API references

Submodules

Actor module

DatabaseActor class for LitePolis.

This class serves as the central point of interaction between the LitePolis system and the database module. It aggregates operations from various manager classes, such as UserManager, ConversationManager, CommentManager, VoteManager, and ReportManager, providing a unified interface for database interactions.

Comments module

This module defines the database schema for comments, including the Comment model and the CommentManager class for managing comment data.

The Comment model represents the comments table and includes fields for the comment’s content (text_field), timestamps (created, modified), moderation status (moderation_status), and foreign keys linking to the user (user_id), conversation (conversation_id), and parent comment (parent_comment_id). It also defines relationships to related User, Conversation, Vote, and other Comment instances (for replies).

The CommentManager class provides static methods for common database operations on comments, such as creating, reading, updating, deleting, listing by various criteria (conversation, user, date range), searching, and counting.

Table Schemas
Table Name	Description
users	Stores user information (id, email, auth_token, etc.).
conversations	Stores conversation information (id, title, etc.).
comments	Stores comment information (id, text_field, user_id, conversation_id, parent_comment_id, created, modified, moderation_status).
votes	Stores vote information (id, user_id, comment_id, value).

Comments Table Details
Column Name	Description
id (int)	Primary key for the comment.
text_field (str)	The content of the comment.
user_id (int, optional)	Foreign key referencing the user who created the comment.
conversation_id (int, optional)	Foreign key referencing the conversation the comment belongs to.
parent_comment_id (int, optional)	Foreign key referencing the parent comment (for replies).
created (datetime)	Timestamp of when the comment was created.
modified (datetime)	Timestamp of when the comment was last modified.
moderation_status (int)	Status indicating the moderation state of the comment (e.g., 0: default, 1: pending, 2: approved, 3: rejected).
user (Relationship)	Relationship to the User who created the comment.
conversation (Relationship)	Relationship to the Conversation the comment belongs to.
votes (Relationship)	Relationship to the Votes associated with this comment.
replies (Relationship)	Relationship to other Comments that are replies to this comment.
parent_comment (Relationship)	Relationship to the parent Comment if this is a reply.

Classes
Class Name	Description
Comment	SQLModel class representing the comments table.
CommentManager	Provides static methods for managing comments (create, read, update, delete, list, search, count).

To use the methods in this module, import DatabaseActor from litepolis_database_default. For example:

from litepolis_database_default import DatabaseActor

comment = DatabaseActor.create_comment({
    "text_field": "This is a test comment.",
    "user_id": 1,
    "conversation_id": 1,
    "moderation_status": 0 # Include new field if needed, or rely on default
})

class litepolis_database_default.Comments.Comment(*, id: int, created: ~datetime.datetime = <factory>, modified: ~datetime.datetime = <factory>, text_field: str, user_id: int | None = None, conversation_id: int | None = None, parent_comment_id: int | None = None, moderation_status: int = 0)

Bases: SQLModel

conversation: Mapped[Conversation | None]

conversation_id: int | None

created: datetime

id: int

model_config: ClassVar[ConfigDict] = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

moderation_status: int

modified: datetime

parent_comment: Mapped[Comment | None]

parent_comment_id: int | None

replies: Mapped[List[Comment]]

text_field: str

user: Mapped[User | None]

user_id: int | None

votes: Mapped[List[Vote]]

class litepolis_database_default.Comments.CommentManager

Bases: object

static count_comments_in_conversation(conversation_id: int) → int

Counts comments in a conversation.

Args:

conversation_id (int): The ID of the conversation to count comments in.

Returns:

int: The number of comments in the given conversation. Returns 0 if no comments are found.

Example:

from litepolis_database_default import DatabaseActor

count = DatabaseActor.count_comments_in_conversation(conversation_id=1)

static create_comment(data: Dict[str, Any]) → Comment

Creates a new Comment record.

Args:

data (Dict[str, Any]): A dictionary containing the data for the new Comment.: Must include ‘text_field’, ‘user_id’, ‘conversation_id’. Can optionally include ‘parent_comment_id’ and ‘moderation_status’.

Returns:

Comment: The newly created Comment instance.

Example:

from litepolis_database_default import DatabaseActor

comment = DatabaseActor.create_comment({
    "text_field": "This is a comment.",
    "user_id": 1,
    "conversation_id": 1,
    "moderation_status": 0
})

static delete_comment(comment_id: int) → bool

Deletes a Comment record by ID.

Args:

comment_id (int): The ID of the Comment to delete.

Returns:

bool: True if the Comment was successfully deleted, False otherwise.

Example:

from litepolis_database_default import DatabaseActor

success = DatabaseActor.delete_comment(comment_id=1)

static get_comment_with_replies(comment_id: int) → Comment | None

Reads a Comment record by ID and loads its direct replies.

Args:

comment_id (int): The ID of the Comment to read.

Returns:

Optional[Comment]: The Comment instance if found, otherwise None.: Direct replies are loaded automatically via the ‘replies’ relationship.

Example:

from litepolis_database_default import DatabaseActor

comment = DatabaseActor.get_comment_with_replies(comment_id=1)
if comment:
    print(f"Comment: {comment.text_field}")
    for reply in comment.replies:
        print(f"- Reply: {reply.text_field}")

static list_comments_by_conversation_id(conversation_id: int, page: int = 1, page_size: int = 10, order_by: str = 'created', order_direction: str = 'asc') → List[Comment]

Lists Comment records for a conversation with pagination and sorting.

Args:

conversation_id (int): The ID of the conversation to list comments for. page (int): The page number to retrieve (default: 1). page_size (int): The number of comments per page (default: 10). order_by (str): The field to order the comments by (default: “created”). Must be a valid column name of the Comment model. order_direction (str): The direction to order the comments in (“asc” or “desc”, default: “asc”).

Returns:

List[Comment]: A list of Comment instances for the given conversation, page, and sorting.

Example:

from litepolis_database_default import DatabaseActor

comments = DatabaseActor.list_comments_by_conversation_id(conversation_id=1, page=1, page_size=10, order_by="created", order_direction="asc")

static list_comments_by_user_id(user_id: int, page: int = 1, page_size: int = 10) → List[Comment]

List comments by user id with pagination.

Args:

user_id (int): The ID of the user to list comments for. page (int): The page number to retrieve (default: 1). page_size (int): The number of comments per page (default: 10).

Returns:

List[Comment]: A list of Comment instances for the given user and page.

Example:

from litepolis_database_default import DatabaseActor

comments = DatabaseActor.list_comments_by_user_id(user_id=1, page=1, page_size=10)

static list_comments_created_in_date_range(start_date: datetime, end_date: datetime) → List[Comment]

List comments created in a date range.

Args:

start_date (datetime): The start date (inclusive) of the range. end_date (datetime): The end date (inclusive) of the range.

Returns:

List[Comment]: A list of Comment instances created within the given date range.

Example:

from litepolis_database_default import DatabaseActor
from datetime import datetime, UTC

start = datetime(2023, 1, 1, tzinfo=UTC)
end = datetime(2023, 1, 31, tzinfo=UTC)
comments = DatabaseActor.list_comments_created_in_date_range(start_date=start, end_date=end)

static read_comment(comment_id: int) → Comment | None

Reads a Comment record by ID.

Args:

comment_id (int): The ID of the Comment to read.

Returns:

Optional[Comment]: The Comment instance if found, otherwise None.

Example:

from litepolis_database_default import DatabaseActor

comment = DatabaseActor.read_comment(comment_id=1)

static search_comments(query: str) → List[Comment]

Search comments by text content using a LIKE query.

Args:

query (str): The search query string. The search will look for: comments where the text_field contains this string.

Returns:

List[Comment]: A list of Comment instances matching the search query.

Example:

from litepolis_database_default import DatabaseActor

comments = DatabaseActor.search_comments(query="search term")

static update_comment(comment_id: int, data: Dict[str, Any]) → Comment | None

Updates a Comment record by ID.

Args:

comment_id (int): The ID of the Comment to update. data (Dict[str, Any]): A dictionary containing the data to update.

Keys should match Comment model field names (e.g., ‘text_field’, ‘moderation_status’).

Returns:

Optional[Comment]: The updated Comment instance if found, otherwise None.

Example:

from litepolis_database_default import DatabaseActor

updated_comment = DatabaseActor.update_comment(comment_id=1, data={"text_field": "Updated comment text.", "moderation_status": 1})

Conversations module

This module defines the Conversation model and ConversationManager class for interacting with the ‘conversations’ table in the database.

The ‘conversations’ table stores information about conversations, including their title, description, archive status, arbitrary settings in a JSON format, and the user who created them. It also includes timestamps for creation and modification.

Table Schema
Column	Type	Description
id	INTEGER	Unique identifier for the conversation.
user_id	INTEGER	Foreign key referencing the user who created the conversation.
title	VARCHAR	Title of the conversation.
description	VARCHAR	Description of the conversation.
is_archived	BOOLEAN	Indicates whether the conversation is archived.
created	DATETIME	Timestamp indicating when the conversation was created.
modified	DATETIME	Timestamp indicating when the conversation was last modified.
settings	JSON	Arbitrary key-value settings for the conversation.

Relationships
comments	One-to-many relationship with the Comment model.
user	Many-to-one relationship with the User model.

The ConversationManager class provides static methods for performing CRUD (Create, Read, Update, Delete) operations on the ‘conversations’ table, as well as methods for listing, searching, counting, and archiving conversations.

To use the methods in this module, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

conversation = DatabaseActor.create_conversation({
    "title": "New Conversation",
    "description": "A new conversation about a topic.",
    "user_id": 1,
    "settings": {"visibility": "public"}
})

class litepolis_database_default.Conversations.Conversation(*, id: int | None = None, user_id: int | None = None, title: str, description: str | None = None, is_archived: bool = False, created: ~datetime.datetime = <factory>, modified: ~datetime.datetime = <factory>, settings: ~typing.Dict[str, ~typing.Any] = {})

Bases: SQLModel

comments: Mapped[List[Comment]]

created: datetime

description: str | None

id: int | None

is_archived: bool

model_config: ClassVar[ConfigDict] = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modified: datetime

settings: Dict[str, Any]

title: str

user: Mapped[User | None]

user_id: int | None

class litepolis_database_default.Conversations.ConversationManager

Bases: object

static archive_conversation(conversation_id: int) → Conversation | None

Archives a conversation.

Sets the ‘is_archived’ field to True for the specified conversation.

Args:

conversation_id (int): The ID of the Conversation to archive.

Returns:

Optional[Conversation]: The archived Conversation instance if found, otherwise None.

Example:

from litepolis_database_default import DatabaseActor

archived_conversation = DatabaseActor.archive_conversation(conversation_id=1)

static count_conversations() → int

Counts all Conversation records.

Returns:

int: The total number of Conversation records.

Example:

from litepolis_database_default import DatabaseActor

count = DatabaseActor.count_conversations()

static create_conversation(data: Dict[str, Any]) → Conversation

Creates a new Conversation record.

Args:

data (Dict[str, Any]): A dictionary containing the data for the new Conversation.: Expected keys include ‘title’ (required), ‘description’ (optional), ‘is_archived’ (optional, defaults to False), ‘user_id’ (optional), and ‘settings’ (optional, defaults to {}).

Returns:

Conversation: The newly created Conversation instance.

Example:

from litepolis_database_default import DatabaseActor

conversation = DatabaseActor.create_conversation({
    "title": "New Conversation",
    "description": "A new conversation about a topic.",
    "user_id": 1,
    "settings": {"visibility": "public"}
})

static delete_conversation(conversation_id: int) → bool

Deletes a Conversation record by ID.

Args:

conversation_id (int): The ID of the Conversation to delete.

Returns:

bool: True if the Conversation was successfully deleted, False otherwise.

Example:

from litepolis_database_default import DatabaseActor

success = DatabaseActor.delete_conversation(conversation_id=1)

static list_conversations(page: int = 1, page_size: int = 10, order_by: str = 'created', order_direction: str = 'desc') → List[Conversation]

Lists Conversation records with pagination and sorting.

Args:

page (int): The page number to retrieve (default: 1). page_size (int): The number of records per page (default: 10). order_by (str): The field to order the results by (default: “created”). order_direction (str): The direction to order the results in (“asc” or “desc”, default: “desc”).

Returns:

List[Conversation]: A list of Conversation instances.

Example:

from litepolis_database_default import DatabaseActor

conversations = DatabaseActor.list_conversations(page=1, page_size=10, order_by="title", order_direction="asc")

static list_conversations_by_archived_status(is_archived: bool) → List[Conversation]

List conversations by archive status.

Args:

is_archived (bool): The archive status to filter by.

Returns:

List[Conversation]: A list of Conversation instances with the specified archive status.

Example:

from litepolis_database_default import DatabaseActor

conversations = DatabaseActor.list_conversations_by_archived_status(is_archived=True)

static list_conversations_created_in_date_range(start_date: datetime, end_date: datetime) → List[Conversation]

List conversations created in date range.

Args:

start_date (datetime): The start date of the range. end_date (datetime): The end date of the range.

Returns:

List[Conversation]: A list of Conversation instances created within the specified date range.

Example:

from litepolis_database_default import DatabaseActor
from datetime import datetime

start = datetime(2023, 1, 1)
end = datetime(2023, 1, 31)
conversations = DatabaseActor.list_conversations_created_in_date_range(start_date=start, end_date=end)

static read_conversation(conversation_id: int) → Conversation | None

Reads a Conversation record by ID.

Args:

conversation_id (int): The ID of the Conversation to read.

Returns:

Optional[Conversation]: The Conversation instance if found, otherwise None.

Example:

from litepolis_database_default import DatabaseActor

conversation = DatabaseActor.read_conversation(conversation_id=1)

static search_conversations(query: str) → List[Conversation]

Search conversations by title or description.

Args:

query (str): The search query.

Returns:

List[Conversation]: A list of Conversation instances that match the search query.

Example:

from litepolis_database_default import DatabaseActor

conversations = DatabaseActor.search_conversations(query="search term")

static update_conversation(conversation_id: int, data: Dict[str, Any]) → Conversation | None

Updates a Conversation record by ID.

Args:

conversation_id (int): The ID of the Conversation to update. data (Dict[str, Any]): A dictionary containing the data to update.

Allowed keys include ‘title’, ‘description’, ‘is_archived’, ‘user_id’, and ‘settings’.

Returns:

Optional[Conversation]: The updated Conversation instance if found, otherwise None.

Example:

from litepolis_database_default import DatabaseActor

updated_conversation = DatabaseActor.update_conversation(conversation_id=1, data={"title": "Updated Title", "settings": {"visibility": "private"}})

Users module

This module defines the User model and UserManager class for interacting with the ‘users’ table in the database.

The ‘users’ table stores information about users, including their email, authentication token, and admin status. It also includes timestamps for creation and modification. The table supports both standard SQL databases and StarRocks, with some specific handling for StarRocks (e.g., unique constraints and primary keys).

Table Schema
Column	Type	Description
id	INTEGER	Unique identifier for the user.
email	VARCHAR	User’s email address. Must be unique.
auth_token	VARCHAR	Authentication token for the user.
is_admin	BOOLEAN	Indicates whether the user is an administrator.
created	DATETIME	Timestamp indicating when the user was created.
modified	DATETIME	Timestamp indicating when the user was last modified.

Relationships
Comment	One-to-many.
Vote	One-to-many.
Conversation	One-to-many.

The UserManager class provides static methods for interacting with the ‘users’ table, including CRUD operations, searching, listing with pagination, and counting.

To use the methods in this module, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

user = DatabaseActor.create_user({
    "email": "test@example.com",
    "auth_token": "auth_token",
})

class litepolis_database_default.Users.User(*, id: int | None, email: str, auth_token: str, is_admin: bool = False, created: ~datetime.datetime = <factory>, modified: ~datetime.datetime = <factory>)

Bases: SQLModel

auth_token: str

comments: Mapped[List[Comment]]

conversation: Mapped[List[Conversation]]

created: datetime

email: str

id: int | None

is_admin: bool

model_config: ClassVar[ConfigDict] = {'from_attributes': True, 'read_from_attributes': True, 'read_with_orm_mode': True, 'registry': PydanticUndefined, 'table': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

modified: datetime

votes: Mapped[List[Vote]]

class litepolis_database_default.Users.UserManager

Bases: object

static count_users() → int

Counts the total number of User records in the database.

Returns:: The total count of users.