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: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, modified: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, text_field: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], user_id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=users.id, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, conversation_id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=conversations.id, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, parent_comment_id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=True, foreign_key=comments.id, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, moderation_status: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = 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, timezone

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

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: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], email: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=True, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], auth_token: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], hname: ~typing.Annotated[str | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, is_admin: ~typing.Annotated[bool, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = False, created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, modified: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>)

Bases: SQLModel

auth_token: str

comments: Mapped[List[Comment]]

conversation: Mapped[List[Conversation]]

created: datetime

email: str

hname: str | None

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.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

count = DatabaseActor.count_users()

static create_user(data: Dict[str, Any]) → User | None

Creates a new User record.

Handles checking for existing email before creation, especially for StarRocks.

Args:

data: A dictionary containing user data (e.g., “email”, “auth_token”).

Returns:

The created User object, or None if a user with the same email already exists.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

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

static delete_user(user_id: int) → bool

Deletes a User record by ID.

Args:

user_id: The unique identifier of the user to delete.

Returns:

True if the user was found and deleted, False otherwise.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

success = DatabaseActor.delete_user(user_id=1)

static list_users(page: int = 1, page_size: int = 10) → List[User]

Lists User records with pagination.

Args:

page: The page number (1-based). Defaults to 1. page_size: The number of records per page. Defaults to 10.

Returns:

A list of User objects for the specified page.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

users = DatabaseActor.list_users(page=1, page_size=10)

static list_users_by_admin_status(is_admin: bool) → List[User]

Lists users based on their admin status.

Args:

is_admin: A boolean indicating whether to list administrators (True) or non-administrators (False).

Returns:

A list of User objects matching the specified admin status.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

users = DatabaseActor.list_users_by_admin_status(is_admin=True)

static list_users_created_in_date_range(start_date: datetime, end_date: datetime) → List[User]

Lists users created within a specified date range (inclusive).

Args:

start_date: The start of the date range. end_date: The end of the date range.

Returns:

A list of User objects created within the specified range.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

users = DatabaseActor.list_users_created_in_date_range(start_date=datetime(2023, 1, 1), end_date=datetime(2023, 12, 31))

static read_user(user_id: int) → User | None

Reads a User record by ID.

Args:

user_id: The unique identifier of the user.

Returns:

The User object if found, otherwise None.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

user = DatabaseActor.read_user(user_id=1)

static read_user_by_email(email: str) → User | None

Reads a User record by email address.

Args:

email: The email address of the user.

Returns:

The User object if found, otherwise None.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

user = DatabaseActor.read_user_by_email(email="test@example.com")

static search_users_by_email(query: str) → List[User]

Searches for users whose email address contains the specified query string.

Args:

query: The string to search for within email addresses.

Returns:

A list of User objects matching the search query.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

users = DatabaseActor.search_users_by_email(query="example.com")

static update_user(user_id: int, data: Dict[str, Any]) → User | None

Updates a User record by ID with the provided data.

Args:

user_id: The unique identifier of the user to update. data: A dictionary containing the fields and new values to update.

Returns:

The updated User object if found and updated, otherwise None.

To use this method, import DatabaseActor. For example:

from litepolis_database_default import DatabaseActor

user = DatabaseActor.update_user(user_id=1, data={"email": "new_email@example.com"})

Vote module

This module defines the database schema for votes, including the Vote model and VoteManager class for managing votes.

The database schema includes tables for users, comments, and votes, with relationships defined between them. The Vote table stores information about individual votes, including the voter, target comment, and vote value.

Table Schemas

Table Name	Description
users	Stores user information (id, email, auth_token, etc.).
comments	Stores comment information (id, text, user_id, conversation_id, parent_comment_id, created, modified).
votes	Stores vote information (id, user_id, comment_id, value, created, modified).

Votes Table Details

Column Name	Description
id (int)	Primary key for the vote.
user_id (int, optional)	Foreign key referencing the user who created the vote.
comment_id (int, optional)	Foreign key referencing the comment being voted on.
value (int)	The value of the vote.
created (datetime)	Timestamp of when the vote was created.
modified (datetime)	Timestamp of when the vote was last modified.

Classes

Class Name	Description
Vote	SQLModel class representing the votes table.
VoteManager	Provides static methods for managing votes.

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

from litepolis_database_default import DatabaseActor

vote = DatabaseActor.create_vote({
    "value": 1,
    "user_id": 1,
    "comment_id": 1
})

class litepolis_database_default.Vote.Vote(*, id: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], value: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], user_id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=users.id, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, comment_id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=comments.id, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, modified: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>)

Bases: SQLModel

comment: Mapped[Comment | None]

comment_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].

modified: datetime

user: Mapped[User | None]

user_id: int | None

value: int

class litepolis_database_default.Vote.VoteManager

Bases: object

static count_votes_for_comment(comment_id: int) → int

Counts votes for a comment.

Args:

comment_id (int): The ID of the comment.

Returns:

int: The number of votes for the comment. Returns 0 if no votes are found for the comment.

Example:

from litepolis_database_default import DatabaseActor

count = DatabaseActor.count_votes_for_comment(comment_id=1)

static create_vote(data: Dict[str, Any]) → Vote

Creates a new Vote record.

Note:

This operation may raise exceptions (e.g., IntegrityError) if: - A vote already exists for the given user_id and comment_id (due to unique constraint). - The user_id or comment_id do not reference existing records in the users or comments tables (due to foreign key constraints).

Args:

data (Dict[str, Any]): A dictionary containing the data for the new Vote.

Must include ‘value’, ‘user_id’, and ‘comment_id’.

Returns:

Vote: The newly created Vote instance.

Example:

from litepolis_database_default import DatabaseActor

vote = DatabaseActor.create_vote({
    "value": 1,
    "user_id": 1,
    "comment_id": 1
})

static delete_vote(vote_id: int) → bool

Deletes a Vote record by ID.

Args:

vote_id (int): The ID of the Vote to delete.

Returns:

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

Returns False if the vote with the given ID does not exist.

Example:

from litepolis_database_default import DatabaseActor

success = DatabaseActor.delete_vote(vote_id=1)

static get_vote_by_user_comment(user_id: int, comment_id: int) → Vote | None

Reads a Vote record by user and comment IDs.

Args:

user_id (int): The ID of the user. comment_id (int): The ID of the comment.

Returns:

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

Example:

from litepolis_database_default import DatabaseActor

vote = DatabaseActor.get_vote_by_user_comment(user_id=1, comment_id=1)

static get_vote_value_distribution_for_comment(comment_id: int) → Dict[int, int]

Gets vote value distribution for a comment.

Args:

comment_id (int): The ID of the comment.

Returns:

Dict[int, int]: A dictionary where the keys are vote values and the values are the counts.

Returns an empty dictionary if no votes are found for the comment.

Example:

from litepolis_database_default import DatabaseActor

distribution = DatabaseActor.get_vote_value_distribution_for_comment(comment_id=1)

static list_votes_by_comment_id(comment_id: int, page: int = 1, page_size: int = 10, order_by: str = 'created', order_direction: str = 'asc') → List[Vote]

Lists Vote records for a comment with pagination and sorting.

Args:

comment_id (int): The ID of the comment. page (int): The page number to retrieve (default: 1). page_size (int): The number of votes per page (default: 10). order_by (str): The field to order the votes by (default: “created”).

Must be a valid attribute name of the Vote model.

order_direction (str): The direction to order the votes (“asc” or “desc”, default: “asc”).

Returns:

List[Vote]: A list of Vote instances. Returns an empty list if no votes are found for the comment or page.

Example:

from litepolis_database_default import DatabaseActor

votes = DatabaseActor.list_votes_by_comment_id(comment_id=1, page=1, page_size=10, order_by="created", order_direction="asc")

static list_votes_by_user_id(user_id: int, page: int = 1, page_size: int = 10) → List[Vote]

List votes by user id with pagination.

Args:

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

Returns:

List[Vote]: A list of Vote instances. Returns an empty list if no votes are found for the user or page.

Example:

from litepolis_database_default import DatabaseActor

votes = DatabaseActor.list_votes_by_user_id(user_id=1, page=1, page_size=10)

static list_votes_created_in_date_range(start_date: datetime, end_date: datetime) → List[Vote]

List votes created in date range.

Args:

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

Returns:

List[Vote]: A list of Vote instances. Returns an empty list if no votes are found in the range.

Example:

from litepolis_database_default import DatabaseActor
from datetime import datetime

start = datetime(2023, 1, 1)
end = datetime(2023, 1, 31)
votes = DatabaseActor.list_votes_created_in_date_range(start_date=start, end_date=end)

static read_vote(vote_id: int) → Vote | None

Reads a Vote record by ID.

Args:

vote_id (int): The ID of the Vote to read.

Returns:

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

Example:

from litepolis_database_default import DatabaseActor

vote = DatabaseActor.read_vote(vote_id=1)

static update_vote(vote_id: int, data: Dict[str, Any]) → Vote | None

Updates a Vote record by ID.

Args:

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

Keys should match Vote model attributes.

Returns:

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

Returns None if the vote with the given ID does not exist.

Example:

from litepolis_database_default import DatabaseActor

updated_vote = DatabaseActor.update_vote(vote_id=1, data={"value": -1})

Participant module

Participant model and ParticipantManager for managing conversation participants.

In Polis, participants are users who have joined a conversation. Each participant has a unique pid per (zid, uid) combination.

Table Schema

Column	Type	Description
pid	INTEGER	Unique participant ID (primary key)
zid	INTEGER	Foreign key to conversation
uid	INTEGER	Foreign key to user
created	DATETIME	When participant joined
modified	DATETIME	Last modification time
vote_count	INTEGER	Number of votes cast
mod	INTEGER	Moderation status (0=active)

To use this module:

from litepolis_database_default import DatabaseActor

participant = DatabaseActor.create_participant({
    "zid": 1,
    "uid": 1
})

class litepolis_database_default.Participant.Participant(*, pid: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = None, zid: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], uid: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, modified: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, vote_count: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = 0, mod: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = 0)

Bases: SQLModel

created: datetime

mod: 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].

modified: datetime

pid: int | None

uid: int

vote_count: int

zid: int

class litepolis_database_default.Participant.ParticipantManager

Bases: object

static count_participants(zid: int) → int

Counts participants in a conversation.

static create_participant(data: Dict[str, Any]) → Participant

Creates a new Participant record.

Args:

data: Must include ‘zid’ and ‘uid’.

Returns:

The created Participant instance.

Example:

from litepolis_database_default import DatabaseActor participant = DatabaseActor.create_participant({“zid”: 1, “uid”: 1})

static delete_participant(pid: int) → bool

Deletes a Participant record.

static get_or_create_anonymous_participant(zid: int, pc_token: str) → Participant

Gets or creates an anonymous participant using a pc token.

Uses negative UIDs derived from hash of pc_token to distinguish anonymous users.

static get_or_create_participant(zid: int, uid: int) → Participant

Gets existing participant or creates new one.

static get_participant_by_zid_uid(zid: int, uid: int) → Participant | None

Gets participant by conversation and user IDs.

static increment_vote_count(pid: int) → Participant | None

Increments vote count for a participant.

static list_participants_by_zid(zid: int, page: int = 1, page_size: int = 100) → List[Participant]

Lists participants in a conversation with pagination.

static read_participant(pid: int) → Participant | None

Reads a Participant by ID.

static update_participant(pid: int, data: Dict[str, Any]) → Participant | None

Updates a Participant record.

Zinvite module

Zinvite model and ZinviteManager for managing conversation invite codes.

Zinvites (zinvite) are short codes that allow users to join conversations. The zinvite code serves as the external conversation_id in the Polis API.

Table Schema

Column	Type	Description
zid	INTEGER	Foreign key to conversation
zinvite	VARCHAR	Unique invite code (primary key)
created	DATETIME	When invite was created

To use this module:

from litepolis_database_default import DatabaseActor

zinvite = DatabaseActor.create_zinvite({
    "zid": 1,
    "zinvite": "abc123def"
})

class litepolis_database_default.Zinvite.Zinvite(*, zid: ~typing.Annotated[int, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], zinvite: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=True, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>)

Bases: SQLModel

created: datetime

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].

zid: int

zinvite: str

class litepolis_database_default.Zinvite.ZinviteManager

Bases: object

static create_zinvite(data: Dict[str, Any]) → Zinvite

Creates a new Zinvite record.

Args:

data: Must include ‘zid’. If ‘zinvite’ not provided, generates one.

Returns:

The created Zinvite instance.

Example:

from litepolis_database_default import DatabaseActor zinvite = DatabaseActor.create_zinvite({“zid”: 1})

static delete_zinvite(zinvite: str) → bool

Deletes a Zinvite record.

static delete_zinvites_by_zid(zid: int) → int

Deletes all zinvites for a conversation. Returns count deleted.

static get_or_create_zinvite(zid: int) → Zinvite

Gets existing zinvite or creates new one for conversation.

static get_zid_by_zinvite(zinvite_code: str) → int | None

Gets zid from zinvite code.

static get_zinvite_by_zid(zid: int) → Zinvite | None

Gets zinvite for a conversation.

static read_zinvite(zinvite: str) → Zinvite | None

Reads a Zinvite by code.

litepolis_database_default.Zinvite.generate_zinvite_code(length: int = 12) → str

Generate a random invite code that starts with a digit.

The Polis frontend router requires conversation IDs to start with a digit, matching the pattern: /^[0-9][0-9A-Za-z]+$/

Einvite module

Einvite model and EinviteManager for managing email invites.

Einivtes are single-use invite codes sent via email that allow users to register or join specific conversations.

Table Schema

Column	Type	Description
einvite	VARCHAR	Unique invite code (primary key)
email	VARCHAR	Email address invite was sent to
created	DATETIME	When invite was created

To use this module:

from litepolis_database_default import DatabaseActor

einvite = DatabaseActor.create_einvite({
    "email": "user@example.com"
})

class litepolis_database_default.Einvite.Einvite(*, einvite: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], email: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>)

Bases: SQLModel

created: datetime

einvite: str

email: str

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].

class litepolis_database_default.Einvite.EinviteManager

Bases: object

static create_einvite(data: Dict[str, Any]) → Einvite

Creates a new Einvite record.

Args:

data: Must include ‘email’. If ‘einvite’ not provided, generates one.

Returns:

The created Einvite instance.

Example:

from litepolis_database_default import DatabaseActor einvite = DatabaseActor.create_einvite({“email”: “user@example.com”})

static delete_einvite(einvite: str) → bool

Deletes an Einvite record (after use).

static delete_einvites_by_email(email: str) → int

Deletes all einvites for an email. Returns count deleted.

static get_einvite_by_email(email: str) → Einvite | None

Gets latest einvite for an email.

static read_einvite(einvite: str) → Einvite | None

Reads an Einvite by code.

static validate_einvite(einvite_code: str, email: str) → bool

Validates an einvite code matches the email.

litepolis_database_default.Einvite.generate_einvite_code(length: int = 16) → str

Generate a random email invite code.

PasswordReset module

Password Reset Token model for handling password reset requests.

class litepolis_database_default.PasswordReset.PasswordResetToken(*, id: ~typing.Annotated[int | None, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], email: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=True, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], token: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=True, index=True, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], created: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, expires: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>, used: ~typing.Annotated[bool, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = False)

Bases: SQLModel

created: datetime

email: str

expires: datetime

id: int | None

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].

token: str

used: bool

class litepolis_database_default.PasswordReset.PasswordResetTokenManager

Bases: object

static cleanup_expired() → int

Remove expired tokens. Returns count of deleted tokens.

static create_token(email: str) → PasswordResetToken

Create a new password reset token for an email address.

static get_valid_token(token: str) → PasswordResetToken | None

Get a valid (not expired, not used) token.

static mark_used(token_id: int) → bool

Mark a token as used.

MigrationRecord module

class litepolis_database_default.MigrationRecord.MigrationRecord(*, id: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=True, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], hash: ~typing.Annotated[str, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=False, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)], executed_at: ~datetime.Annotated[~datetime.datetime, ~sqlmodel.main.FieldInfoMetadata(primary_key=PydanticUndefined, nullable=PydanticUndefined, foreign_key=PydanticUndefined, ondelete=PydanticUndefined, unique=PydanticUndefined, index=PydanticUndefined, sa_type=PydanticUndefined, sa_column=PydanticUndefined, sa_column_args=PydanticUndefined, sa_column_kwargs=PydanticUndefined)] = <factory>)

Bases: SQLModel

executed_at: datetime

hash: str

id: str

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].

class litepolis_database_default.MigrationRecord.MigrationRecordManager

Bases: object

static create_migration(data: Dict[str, Any]) → MigrationRecord

Creates a new MigrationRecord record.

static delete_migration(migration_id: str) → bool

Deletes a MigrationRecord record by ID.

static get_latest_executed_migration() → MigrationRecord | None

Returns the latest executed migration record.

static list_executed_migrations(page: int = 1, page_size: int = 10, order_by: str = 'executed_at', order_direction: str = 'desc') → List[MigrationRecord]

Lists executed MigrationRecord records with pagination and sorting.

static read_migration(migration_id: str) → MigrationRecord | None

Reads a MigrationRecord record by ID.

static verify_migration_integrity(migration_id: str, file_content: bytes) → bool

Verifies migration file integrity by comparing hashes.

utils module

litepolis_database_default.utils.connect_db()

Initialize database connection. Engine is already created globally.

litepolis_database_default.utils.get_session()

litepolis_database_default.utils.is_starrocks_engine(engine=Engine(sqlite:///database.db)) → bool

Determine if the engine is connected to StarRocks

litepolis_database_default.utils.wait_for_alter_completion(conn, table_name: str, timeout=30)

Wait for the latest ongoing schema change for a specific table to complete using ‘ORDER BY CreateTime DESC LIMIT 1’.

utils_StarRocks module

litepolis_database_default.utils_StarRocks.create_db_and_tables()

Creates all tables, using standard or custom DDL based on dialect. Relies on models being imported beforehand.

litepolis_database_default.utils_StarRocks.generate_custom_ddl_for_table(table: Table, hints: Dict[str, Any], dialect: Dialect)

Generates the specific DDL for one table using hints and the provided dialect. Handles column order and key placement.

litepolis_database_default.utils_StarRocks.get_hints_for_table(table_name: str) → Dict[str, Any] | None

Retrieves hints from the registry based on table name.

litepolis_database_default.utils_StarRocks.populate_registry(metadata: MetaData)

Populates the _MODEL_REGISTRY using table names from metadata after all decorated models are imported and defined.

litepolis_database_default.utils_StarRocks.register_table(distributed_by: str | None = None, properties: Dict[str, str] | None = None)

Decorator to register a SQLModel class and its special DDL hints for custom DDL generation.

Module contents