lback.repositories package

This file serves as the initialization point for the ‘lback_framework/lback/repositories’ package. It is designed to expose the core components necessary for implementing the repository pattern within the Lback web framework. This package centralizes the definition and management of data access logic, providing an abstraction layer over the database for specific models. It encapsulates the mechanisms for retrieving, storing, and querying data, promoting a clean separation of concerns between the application’s business logic and data persistence.

—

Key Components Exposed by this Package:

1. AdminUserRepository (from .admin_user_repository): A repository class specifically designed for interacting with AdminUser data. This component provides methods for performing CRUD (Create, Read, Update, Delete) operations on administrative user records, abstracting away the direct database interactions for AdminUser models. It might include specialized queries relevant to admin user management.

2. PermissionRepository (from .permission_repository): A repository class dedicated to managing Permission data. This component offers methods for accessing, creating, updating, and deleting permission records. It ensures that permission-related database operations are handled consistently and efficiently, supporting the application’s authorization system.

3. RoleRepository (from .role_repository): A repository class focused on handling Role data. This component provides methods for performing CRUD operations on role records, which are used to group permissions and assign them to users. It centralizes the data access logic for roles, supporting the role-based access control (RBAC) system.

4. UserRepository (from .user_repository): A repository class specifically designed for interacting with standard User data. This component provides methods for performing CRUD operations on general user records, abstracting away the direct database interactions for User models. It might include specialized queries for user authentication, profile management, or other user-centric operations.

Submodules

lback.repositories.admin_user_repository module

class lback.repositories.admin_user_repository.AdminUserRepository(session: Session)

Bases: object

Repository for AdminUser model data access. Provides methods to interact with the database for AdminUser entities. Requires a SQLAlchemy Session to be provided upon initialization. Integrates SignalDispatcher to emit events related to repository operations.

__init__(session: Session)

Initializes the AdminUserRepository with a SQLAlchemy session.

Parameters:

session – The SQLAlchemy Session object to use for database operations. This session should be managed externally (e.g., by a middleware).

create(**data: Any) → AdminUser

Creates a new admin user instance and adds it to the session. Note: This method adds the object to the session but does NOT commit. The caller is responsible for committing the session. Emits ‘admin_user_pre_create’ and ‘admin_user_post_create’ signals.

Parameters:

**data – Keyword arguments corresponding to AdminUser model attributes. Example: username=’admin’, email=’a@example.com’, password=’hashed_password’.

Returns:

The newly created AdminUser instance (staged in the session).

Raises:

Exception – If an error occurs during instance creation or adding to session.

delete(admin_user: AdminUser)

Deletes an admin user instance from the session. Note: This method marks the object for deletion but does NOT commit. The caller is responsible for committing the session. Emits ‘admin_user_pre_delete’ and ‘admin_user_post_delete’ signals.

Parameters:

admin_user – The AdminUser instance to delete.

Raises:

Exception – If an error occurs during the deletion process.

get_admin_by_reset_token(token: str) → AdminUser | None

Gets an admin user by a password reset token. # Assuming this method exists and works. No signals added here, # as token validation/lookup signals might be handled elsewhere (e.g., UserManager).

get_by_email(email: str) → AdminUser | None

Gets an admin user by their email address. # No signals here, as this is a simple read operation.

Parameters:

email – The email address string of the admin user.

Returns:

The AdminUser object if found, otherwise None.

get_by_id(admin_user_id: int) → AdminUser | None

Gets an admin user by their primary key ID. # No signals here, as this is a simple read operation.

Parameters:

admin_user_id – The integer ID of the admin user.

Returns:

The AdminUser object if found, otherwise None.

get_by_username(username: str) → AdminUser | None

Gets an admin user by their username. # No signals here, as this is a simple read operation.

Parameters:

username – The username string of the admin user.

Returns:

The AdminUser object if found, otherwise None.

list_all() → List[AdminUser]

Lists all admin users in the database. # No signals here, as this is a simple read operation.

Returns:

A list of all AdminUser objects.

search(**criteria: Any) → List[AdminUser]

Searches for admin users based on criteria. # No signals here, as this is a read operation.

update(admin_user: AdminUser, **data: Any) → AdminUser

Updates an existing admin user instance with new data. Note: This method modifies the object in the session but does NOT commit. The caller is responsible for committing the session. Emits ‘admin_user_pre_update’ and ‘admin_user_post_update’ signals.

Parameters:

• admin_user – The AdminUser instance to update.

• **data – Keyword arguments for the attributes to update. Example: email=’new_email@example.com’, is_active=False.

Returns:

The updated AdminUser instance (changes staged in the session).

Raises:

Exception – If an error occurs during the update process.

lback.repositories.permission_repository module

class lback.repositories.permission_repository.PermissionRepository(session: Session)

Bases: object

Repository for Permission model data access. Provides methods to interact with the database for Permission entities. Requires a SQLAlchemy Session to be provided upon initialization. Integrates SignalDispatcher to emit events related to repository operations.

__init__(session: Session)

Initializes the PermissionRepository with a SQLAlchemy session.

Parameters:

session – The SQLAlchemy Session object to use for database operations. This session should be managed externally (e.g., by a middleware).

create(**data: Any) → Permission

Creates a new permission instance and adds it to the session. Note: This method adds the object to the session but does NOT commit. The caller is responsible for committing the session. Emits ‘permission_pre_create’ and ‘permission_post_create’ signals.

Parameters:

**data – Keyword arguments corresponding to Permission model attributes. Example: name=’add_product’, description=’Can add new products’.

Returns:

The newly created Permission instance (staged in the session).

Raises:

Exception – For errors during instance creation or adding to session.

delete(permission: Permission)

Deletes a permission instance from the session. Note: This method marks the object for deletion but does NOT commit. The caller is responsible for committing the session. Emits ‘permission_pre_delete’ and ‘permission_post_delete’ signals.

Parameters:

permission – The Permission instance to delete.

Raises:

Exception – For errors during marking for deletion.

get_by_id(permission_id: int) → Permission | None

Gets a permission by its primary key ID. # No signals here, as this is a simple read operation.

Parameters:

permission_id – The integer ID of the permission.

Returns:

The Permission object if found, otherwise None.

get_by_name(name: str) → Permission | None

Gets a permission by its name. # No signals here, as this is a simple read operation.

Parameters:

name – The name string of the permission.

Returns:

The Permission object if found, otherwise None.

list_all() → List[Permission]

Lists all permissions in the database. # No signals here, as this is a simple read operation.

Returns:

A list of all Permission objects.

search(**criteria: Any) → List[Permission]

Searches for permissions based on criteria. # No signals here, as this is a read operation.

update(permission: Permission, **data: Any) → Permission

Updates an existing permission instance with new data. Note: This method modifies the object in the session but does NOT commit. The caller is responsible for committing the session. Emits ‘permission_pre_update’ and ‘permission_post_update’ signals.

Parameters:

• permission – The Permission instance to update.

• **data – Keyword arguments for the attributes to update. Example: description=’Updated description’.

Returns:

The updated Permission instance (changes staged in the session).

Raises:

Exception – For errors during attribute update.

lback.repositories.role_repository module

class lback.repositories.role_repository.RoleRepository(session: Session)

Bases: object

Repository for Role model data access. Provides methods to interact with the database for Role entities. Requires a SQLAlchemy Session to be provided upon initialization. Integrates SignalDispatcher to emit events related to repository operations.

__init__(session: Session)

Initializes the RoleRepository with a SQLAlchemy session.

Parameters:

session – The SQLAlchemy Session object to use for database operations. This session should be managed externally (e.g., by a middleware).

create(**data: Any) → Role

Creates a new role instance and adds it to the session. Note: This method adds the object to the session but does NOT commit. The caller is responsible for committing the session. Emits ‘role_pre_create’ and ‘role_post_create’ signals.

Parameters:

**data – Keyword arguments corresponding to Role model attributes. Example: name=’editor’, description=’Can edit content’.

Returns:

The newly created Role instance (staged in the session).

Raises:

Exception – For errors during instance creation or adding to session.

delete(role: Role)

Deletes a role instance from the session. Note: This method marks the object for deletion but does NOT commit. The caller is responsible for committing the session. Emits ‘role_pre_delete’ and ‘role_post_delete’ signals.

Parameters:

role – The Role instance to delete.

Raises:

Exception – For errors during marking for deletion.

get_by_id(role_id: int) → Role | None

Gets a role by its primary key ID. # No signals here, as this is a simple read operation.

Parameters:

role_id – The integer ID of the role.

Returns:

The Role object if found, otherwise None.

get_by_name(name: str) → Role | None

Gets a role by its name. # No signals here, as this is a simple read operation.

Parameters:

name – The name string of the role.

Returns:

The Role object if found, otherwise None.

list_all() → List[Role]

Lists all roles in the database. # No signals here, as this is a simple read operation.

Returns:

A list of all Role objects.

search(**criteria: Any) → List[Role]

Searches for roles based on criteria. # No signals here, as this is a read operation.

update(role: Role, **data: Any) → Role

Updates an existing role instance with new data. Note: This method modifies the object in the session but does NOT commit. The caller is responsible for committing the session. Emits ‘role_pre_update’ and ‘role_post_update’ signals.

Parameters:

• role – The Role instance to update.

• **data – Keyword arguments for the attributes to update. Example: description=’Updated description’.

Returns:

The updated Role instance (changes staged in the session).

Raises:

Exception – For errors during attribute update.

lback.repositories.user_repository module

class lback.repositories.user_repository.UserRepository(session: Session)

Bases: object

Repository for User model data access. Provides methods to interact with the database for User entities. Requires a SQLAlchemy Session to be provided upon initialization. Integrates SignalDispatcher to emit events related to repository operations.

__init__(session: Session)

Intializes the UserRepository with a SQLAlchemy session.

Parameters:

session – The SQLAlchemy Session object to use for database operations. This session should be managed externally (e.g., by a middleware).

create(**data: Any) → User

Creates a new user instance and adds it to the session. Note: This method adds the object to the session but does NOT commit. The caller is responsible for committing the session. Emits ‘user_pre_create’ and ‘user_post_create’ signals.

delete(user: User)

Deletes a user instance from the session. Note: This method marks the object for deletion but does NOT commit. The caller is responsible for committing the session. Emits ‘user_pre_delete’ and ‘user_post_delete’ signals.

get_by_email(email: str) → User | None

Gets a user by their email address.

get_by_email_verification_token(token: str) → User | None

Gets a user by their email verification token. Note: The expiry check for email verification token is handled within the User model’s

verify_email method to allow for more flexible expiry logic. This method only retrieves the user based on the token itself.

Parameters:

token – The email verification token.

Returns:

The User object if found, otherwise None.

get_by_id(user_id: int) → User | None

Gets a user by their primary key ID.

get_by_username(username: str) → User | None

Gets a user by their username.

get_user_by_auth_token(token: str) → User | None

Gets a user by an authentication token (e.g., API token). Note: This is different from the reset/verification token as it might not have an expiry. If your ‘auth_token’ is ONLY for password reset, use get_user_by_auth_token_and_expiry. If it’s for API tokens (persistent), then this method would be relevant.

get_user_by_auth_token_and_expiry(token: str) → User | None

Gets a user by their authentication token (e.g., password reset token) and ensures the token has not expired.

Parameters:

token – The authentication token.

Returns:

The User object if found and token is valid/not expired, otherwise None.

get_user_by_reset_token(token: str) → User | None

Gets a user by a password reset token. This method is now a wrapper for get_user_by_auth_token_and_expiry.

list_all() → List[User]

Lists all users in the database.

search(**criteria: Any) → List[User]

Searches for users based on criteria.

update(user: User, **data: Any) → User

Updates an existing user instance with new data. Note: This method modifies the object in the session but does NOT commit. The caller is responsible for committing the session. Emits ‘user_pre_update’ and ‘user_post_update’ signals.