kamihi.db

Database connections module for the Kamihi framework.

License

MIT

Modules:

Name	Description
db	Database connection module for the Kamihi framework.
models	Internal models for Kamihi.

Classes:

Name	Description
BaseUser	Base class for user models.
Job	Model for scheduled jobs.
Pages	Model for pages in a paginated media type.
Permission	Model for permissions.
RegisteredAction	Model for registered actions.
Role	Model for roles.

Functions:

Name	Description
get_engine	Create a database engine.
init_engine	Initialize the database engine.

BaseUser

Bases: BaseModel

Base class for user models.

This class should be extended in user code to create a custom user model.

Attributes:

Name	Type	Description
id	int	Primary key.
telegram_id	int	Unique Telegram ID of the user.
is_admin	bool	Whether the user is an admin.
roles	list[Role]	List of roles associated with the user.
permissions	list[Permission]	List of permissions associated with the user.

Methods:

Name	Description
admin_repr	Define the representation of the user in the admin interface.
cls	Get the active user class.

admin_repr

admin_repr() -> str

Define the representation of the user in the admin interface.

Source code in src/kamihi/db/models.py

def admin_repr(self) -> str:
    """Define the representation of the user in the admin interface."""
    return str(self.telegram_id)

cls classmethod

cls() -> type[BaseUser]

Get the active user class.

Source code in src/kamihi/db/models.py

@classmethod
def cls(cls) -> type[BaseUser]:
    """Get the active user class."""
    return cls._active_class or globals()["User"]

Job

Bases: BaseUuidModel

Model for scheduled jobs.

Attributes:

Name	Type	Description
id	int	Primary key.
action_id	int	Foreign key to the registered action.
action	RegisteredAction	The registered action associated with the job.
cron_expression	str	Cron expression defining the job schedule.
enabled	bool	Whether the job is active.
args	dict	Arguments to pass to the job when executed.
users	list[User]	List of users associated with the job.

effective_users property

effective_users: list[BaseUser]

Get the list of users associated with this job, either directly or through roles.

Returns:

Type	Description
list[BaseUser]	list[User]: List of users associated with this job.

Pages

Bases: BaseUuidModel

Model for pages in a paginated media type.

Attributes:

Name	Type	Description
id	int	Primary key.

Permission

Bases: BaseModel

Model for permissions.

Attributes:

Name	Type	Description
id	int	Primary key.
action_id	int | None	Foreign key to the registered action.
action	RegisteredAction	The registered action associated with the permission.
users	list[User]	List of users associated with the permission.
roles	list[Role]	List of roles associated with the permission.

Methods:

Name	Description
is_user_allowed	Check if a user has this permission.

effective_users property

effective_users: list[BaseUser]

Get the list of users who have this permission, either directly or through roles.

Returns:

Type	Description
list[BaseUser]	list[User]: List of users with this permission.

is_user_allowed

is_user_allowed(user: BaseUser) -> bool

Check if a user has this permission.

Parameters:

Name	Type	Description	Default
user	User	The user to check.	required

Returns:

Name	Type	Description
bool	bool	True if the user has this permission, False otherwise.

Source code in src/kamihi/db/models.py

def is_user_allowed(self, user: BaseUser) -> bool:
    """
    Check if a user has this permission.

    Args:
        user (User): The user to check.

    Returns:
        bool: True if the user has this permission, False otherwise.

    """
    return user in self.effective_users

RegisteredAction

Bases: BaseModel

Model for registered actions.

Attributes:

Name	Type	Description
id	int	Primary key.
name	str	Name of the action.
description	str | None	Description of the action.
permissions	list[Permission]	List of permissions associated with the action.

Role

Bases: BaseModel

Model for roles.

Attributes:

Name	Type	Description
id	int	Primary key.
name	str	Name of the role.
users	list[User]	List of users associated with the role.
permissions	list[Permission]	List of permissions associated with the role.

get_engine

get_engine() -> Engine

Create a database engine.

Returns:

Name	Type	Description
Engine	Engine	The database engine.

Source code in src/kamihi/db/db.py

def get_engine() -> Engine:
    """
    Create a database engine.

    Returns:
        Engine: The database engine.

    """
    if _engine is None:
        raise RuntimeError("Database engine is not initialized. Call init_engine() first.")
    return _engine

init_engine

init_engine() -> None

Initialize the database engine.

Source code in src/kamihi/db/db.py

def init_engine() -> None:
    """Initialize the database engine."""
    global _engine  # skipcq: PYL-W0603

    db_settings = get_settings().db

    if _engine is None:
        _engine = create_engine(db_settings.url)