lback.utils package

This file serves as the initialization point for the ‘lback_framework/lback/utils’ package. It is designed to expose a collection of various utility functions and helper classes that are commonly used across different parts of the Lback web framework. This package centralizes general-purpose functionalities that do not fit neatly into other specific framework components, enhancing code reusability and simplifying common development tasks.

Key Components Exposed by this Package:

  1. AdminUserManager (from .admin_user_manager): A utility class specifically designed for managing administrative user accounts. This component provides helper methods for common operations related to admin users, such as creating, retrieving, updating, or deleting admin user records, potentially integrating with the AdminUserRepository.

  2. AppSession (from .app_session): A utility or class for managing application-specific session data. This might be a higher-level abstraction over the core session management, providing convenient methods for storing and retrieving data relevant to the current user’s application session.

  3. EmailSender (from .email_sender): A utility class or function for sending emails from the application. This component abstracts away the complexities of configuring and interacting with email servers, providing a simple interface for sending various types of emails (e.g., password resets, notifications).

  4. File Handlers (from .file_handlers): A set of utility functions for handling file operations, particularly related to uploaded files.

    • validate_uploaded_file: A function to validate properties of an uploaded file,

      such as its size, type, or allowed extensions, before saving.

    • save_uploaded_file: A function to securely save an uploaded file to a specified

      location on the server’s file system.

    • delete_saved_file: A function to remove a previously saved file from the server.

  5. Filters (from .filters): A collection of utility functions that can be used for filtering or transforming data.

    • file_extension_filter: A filter to process or validate file extensions.

    • split_filter: A filter to split strings based on a delimiter.

    • date_filter: A filter to format or parse date values.

  6. Response Helpers (from .response_helpers): Utility functions that simplify the creation of common HTTP responses.

    • json_response: A helper function to quickly create and return an HTTP response

      with JSON content, setting appropriate headers.

  7. SessionManager (from .session_manager): A utility class for lower-level management of user sessions. This component might handle the creation, retrieval, updating, and deletion of session data, interacting directly with the session storage mechanism.

  8. Shortcuts (from .shortcuts): A collection of convenience functions designed to simplify common tasks in views or other parts of the application, reducing boilerplate code.

    • render: A shortcut function to render an HTML template and return an HTMLResponse.

    • redirect: A shortcut function to create an HTTP redirect response.

    • return_403: A shortcut to return an HTTP 403 Forbidden response.

    • return_404: A shortcut to return an HTTP 404 Not Found response.

    • return_500: A shortcut to return an HTTP 500 Internal Server Error response.

    • _get_model_form_data: A private or internal helper to extract form data related to a model.

    • paginate_query: A helper to paginate a database query result.

    • json_response: (Duplicated, likely for direct import convenience) A shortcut to return a JSON response.

  9. Static Files (from .static_files): Utilities for managing and serving static assets.

    • static: A function to generate URLs for static files.

    • find_static_file: A function to locate a static file within the project’s static directories.

  10. URL Utilities (from .urls): Utilities related to URL pattern definition.

    • path: A function for defining individual URL routes, mapping a URL pattern to a view.

  11. UserManager (from .user_manager): A utility class specifically designed for managing standard user accounts. This component provides helper methods for common operations related to users, such as creating, retrieving, updating, or deleting user records, potentially integrating with the UserRepository.

  12. Validation (from .validation): Components for general data validation beyond form-specific validation.

    • ValidationError: A custom exception or class used to signal that validation has failed.

    • PasswordValidator: A utility class or function for validating password strength and complexity.

    • validate_json: A function to validate if a given string is valid JSON or conforms to a schema.

Submodules

lback.utils.admin_user_manager module

class lback.utils.admin_user_manager.AdminUserManager[source][source]

Bases: object

Service layer for Admin User related business logic. Manages workflows like registration and authentication. Receives the request-scoped database session per method call and instantiates Repositories with that session. Integrates SignalDispatcher to emit events related to admin user management.

__init__()[source][source]

Initializes the AdminUserManager. Repositories are instantiated per method call with the request-scoped database session provided to that method.

authenticate_admin(session: Session, username: str, password: str) AdminUser | None[source][source]

Authenticates an admin user by username and password. Verifies the plain text password against the stored hash using PasswordHasher. Receives the request-scoped database session. Emits ‘admin_authentication_started’, ‘admin_authenticated’, or ‘admin_authentication_failed’ signals.

Parameters:
  • session – The SQLAlchemy Session for database operations.

  • username – The username to authenticate.

  • password – The plain text password.

Returns:

The AdminUser object if authentication is successful and user is active, otherwise None.

get_admin_by_id(session: Session, admin_id: int) AdminUser | None[source][source]

Get an admin user by ID using the repository. Receives the request-scoped database session. # Consider adding signals if needed, e.g., ‘admin_fetched_by_id’, ‘admin_not_found_by_id’

get_admin_by_username(session: Session, username: str) AdminUser | None[source][source]

Get an admin user by username using the repository. Receives the request-scoped database session. # Consider adding signals if needed, e.g., ‘admin_fetched_by_username’, ‘admin_not_found_by_username’

get_all_admins(session: Session) List[AdminUser][source][source]

Get all admin users using the repository. Receives the request-scoped database session. # Consider adding signals if needed, e.g., ‘all_admins_fetched’

static has_permission(admin_user: AdminUser | None, permission_name: str) bool[source][source]

Check if the admin user has a specific permission. This is a static method as it operates on the user object, not the manager instance. Permission check signals are typically handled by the PermissionRequired decorator.

permission_required() Callable[source][source]

Decorator to check if the authenticated user has a specific permission. Assumes the authenticated user object is available on request.user. Uses the static has_permission method. Permission check signals are handled by the PermissionRequired decorator logic itself.

register_admin(session: Session, username: str, email: str, password: str, role_name: str | None = None) AdminUser | None[source][source]

Registers a new admin user. Handles validation and password hashing before creating via repository. Receives the request-scoped database session. Emits ‘admin_registration_started’, ‘admin_pre_register’, ‘admin_registered’, or ‘admin_registration_failed’ signals.

Parameters:
  • session – The SQLAlchemy Session for database operations.

  • username – The username for the new admin.

  • email – The email for the new admin.

  • password – The plain text password for the new admin.

  • role_name – Optional name of the role to assign to the admin user.

Returns:

The newly created AdminUser object (added to session, but not committed) if successful, otherwise None.

Raises:
  • ValueError – If input data is invalid (missing fields, invalid email, username/email exists, role not found).

  • RuntimeError – If password hashing fails.

  • Exception – For other unexpected errors during database interaction.

lback.utils.app_session module

class lback.utils.app_session.AppSession(session_id: str | None, initial_data: Dict[str, Any] | None, session_manager: SessionManager, db_session: Session, is_new: bool = False, expires_at: datetime | None = None)[source][source]

Bases: object

Represents the session data for a single request, stored in the database. Provides a dictionary-like interface and interacts with the SessionManager.

__contains__(key: Any) bool[source][source]

Checks if a key exists in the session data (e.g., ‘user_id’ in session).

__delitem__(key: str)[source][source]

Deletes a key from the session data using item deletion (e.g., del session[‘user_id’]).

__getitem__(key: str) Any[source][source]

Gets a value from the session data using item access (e.g., session[‘user_id’]).

__init__(session_id: str | None, initial_data: Dict[str, Any] | None, session_manager: SessionManager, db_session: Session, is_new: bool = False, expires_at: datetime | None = None)[source][source]

Initializes the Session wrapper. :param session_id: The ID of the session (None if it’s a new, unsaved session). :param initial_data: The dictionary containing the actual session ‘data’ (or None for a new session).

This data is typically loaded from the DB by SessionMiddleware.

Parameters:
  • session_manager – The SessionManager instance used for saving/deleting.

  • db_session – The SQLAlchemy Session for database operations for this request.

  • is_new – Boolean indicating if this is a newly created session that needs saving and a cookie.

  • expires_at – The datetime when the session expires, loaded from the database.

__iter__()[source][source]

Returns an iterator for the session data keys.

__len__() int[source][source]

Returns the number of items in the session data.

__repr__() str[source][source]

Provides a developer-friendly string representation of the Session wrapper.

__setitem__(key: str, value: Any)[source][source]

Sets a value in the session data using item assignment (e.g., session[‘user_id’] = 123).

__str__() str[source][source]

Provides a user-friendly string representation of the session data.

clear()[source][source]

Clears all data from the session.

delete()[source][source]

Deletes the session using the SessionManager and marks the local object as deleted.

property deleted: bool

Returns True if the session has been marked for deletion.

get(key: str, default: Any | None = None) Any | None[source][source]

Gets a value from the session data using the get() method.

get_flashed_messages(category_filter: str | None = None) list[Dict[str, str]][source][source]

Retrieves all flashed messages from the session and clears them. Optionally filters messages by category.

invalidate()[source][source]

Marks the session for deletion.

property is_new: bool

Returns True if this is a newly created session.

items()[source][source]

Returns a view object that displays a list of a dictionary’s key-value tuple pairs.

keys()[source][source]

Returns a view object that displays a list of all the keys in the dictionary.

load() bool[source][source]

Loads session data from the database using the SessionManager. Returns True if a session was loaded, False otherwise.

property modified: bool

Returns True if the session data has been modified during the request.

pop(key: str, default: Any | None = None) Any | None[source][source]

Removes a key and returns its value.

save() str | None[source][source]

Saves the current session data to the database using the SessionManager. If it’s a new session, it will be created. If modified, it will be updated. :returns: The session ID if save was successful, None otherwise.

property session_id: str | None

Returns the session ID, or None if it’s a new unsaved session.

set_flash(message: str, category: str = 'info')[source][source]

Adds a flash message to the session. Flash messages are typically displayed once on the next request and then cleared.

values()[source][source]

Returns a view object that displays a list of all the values in the dictionary.

lback.utils.email_sender module

class lback.utils.email_sender.EmailSender(smtp_server: str, smtp_port: int, smtp_username: str, smtp_password: str, sender_email: str, use_tls: bool = True, sender_name: str = 'Your App Name')[source][source]

Bases: object

Utility class for sending emails. Configured to use parameters passed during initialization for security and flexibility.

__init__(smtp_server: str, smtp_port: int, smtp_username: str, smtp_password: str, sender_email: str, use_tls: bool = True, sender_name: str = 'Your App Name')[source][source]

Initializes the EmailSender with SMTP configuration.

Parameters:
  • smtp_server – The SMTP server address (e.g., ‘smtp.example.com’).

  • smtp_port – The SMTP server port (e.g., 587).

  • smtp_username – The username for SMTP authentication.

  • smtp_password – The password for SMTP authentication.

  • sender_email – The email address to use as the sender.

  • use_tls – Boolean indicating whether to use TLS encryption (default is True).

  • sender_name – The name to display as the sender (default is “Your App Name”).

send_email(to_email: str, subject: str, body: str, is_html: bool = False)[source][source]

Internal method to handle the actual email sending process. Uses instance attributes for SMTP configuration.

send_generic_email(to_email: str, subject: str, body: str, is_html: bool = False) bool[source][source]

Sends a generic email (plain text or HTML).

send_password_reset_email(to_email: str, username: str, reset_link: str) bool[source][source]

Sends a password reset link to a user.

send_verification_email(to_email: str, username: str, verification_link: str) bool[source][source]

Sends an email verification link to a new user.

lback.utils.file_handlers module

lback.utils.file_handlers.delete_saved_file(relative_file_path: str, config: Config) bool[source][source]

Deletes a saved file from the server’s file system based on its relative path. Returns True if successful, False otherwise.

lback.utils.file_handlers.save_uploaded_file(uploaded_file: UploadedFile, config: Config, model_name: str, allowed_types: List[str] | None = None, max_size_mb: int | None = None) str | None[source][source]

Saves an uploaded file to the server’s file system. Determines save path based on config and model name. Returns the relative path to the saved file, or None on failure.

lback.utils.file_handlers.validate_uploaded_file(uploaded_file: UploadedFile, allowed_types: List[str] | None = None, max_size_mb: int | None = None) str | None[source][source]

Validates an uploaded file based on allowed types and maximum size.

Parameters:
  • uploaded_file – The UploadedFile object.

  • allowed_types – A list of allowed MIME types (e.g., [‘image/jpeg’, ‘image/png’]). If None, all types are allowed.

  • max_size_mb – Maximum allowed file size in megabytes. If None, no size limit.

Returns:

None if validation succeeds, otherwise a string with an error message.

lback.utils.filters module

lback.utils.filters.date_filter(value, format_string='%Y')[source][source]

Jinja2 filter to format datetime objects. By default, it formats to the year (%Y).

Example usage in Jinja2: {{ “now” | date(“%Y”) }} {{ some_datetime_object | date(“%Y-%m-%d”) }}

lback.utils.filters.file_extension_filter(value)[source][source]

Custom Jinja2 filter to extract file extension from a string. Returns lowercase extension or empty string if not applicable.

lback.utils.filters.split_filter(value, sep=None)[source][source]

Custom Jinja2 filter to split a string into a list. Usage in template: {{ value | split(‘.’) }}

lback.utils.response_helpers module

lback.utils.response_helpers.json_response(data: Any, status: int = 200, headers: Dict[str, str] | None = None) Response[source][source]

Helper to return JSON responses with appropriate headers.

lback.utils.session_manager module

class lback.utils.session_manager.SessionManager(timeout_minutes: int = 30)[source][source]

Bases: object

Manages server-side sessions using a persistent database store. Receives the request-scoped database session per method call. Integrates SignalDispatcher to emit events related to session lifecycle and operations.

__init__(timeout_minutes: int = 30)[source][source]

Initializes the SessionManager. Emits ‘session_manager_initialized’ signal.

Parameters:

timeout_minutes – The duration in minutes after which a session expires if not renewed. Defaults to 30 minutes.

__repr__() str[source][source]

Provides a developer-friendly string representation of the SessionManager.

cleanup_sessions(db_session: Session)[source][source]

Removes all expired sessions from the database. This method should be called periodically (e.g., by a background task). Emits ‘session_cleanup_started’ and ‘session_cleanup_finished’ signals.

create_session(db_session: Session, user_id: str | None = None) str | None[source][source]

Creates a new session record in the database and returns its ID.

This method generates a unique session ID, sets its creation and expiration timestamps, and initializes its data payload as an empty JSON string. The session is flushed to the database to ensure the ID is available immediately.

Parameters:
  • db_session – The SQLAlchemy Session for database operations.

  • user_id – An optional user ID to associate with the session upon creation. This value will be stored directly in the user_id column of the Session model.

Returns:

The newly generated unique session ID (string) if successful, otherwise None.

delete_session(db_session: Session, session_id: str) bool[source][source]

Deletes a specific session from the database. This is typically used during user logout or when a session needs to be invalidated.

Parameters:
  • db_session – The SQLAlchemy database session object.

  • session_id – The unique ID of the session to delete.

Returns:

True if the session was found and successfully deleted, False otherwise.

generate_csrf_token(db_session: Session, session_id: str) str | None[source][source]

Generates a new CSRF token, stores it in the session data in the database, and returns it. Emits ‘csrf_token_generated’ signal on success. Emits ‘session_operation_failed’ signal on failure (session not found/expired).

Parameters:
  • db_session – The SQLAlchemy Session for database operations.

  • session_id – The ID of the session to associate the token with.

Returns:

The generated CSRF token string if the session is valid, otherwise None.

get(request: Request, db_session: Session, key: str, default: Any | None = None) Any | None[source][source]

Retrieves a specific value from the session data based on the request and key, using the database. This method extracts the session ID from the request (e.g., from a cookie) and then retrieves the value from the session data in the database. Relies on get_session_data which emits signals.

Parameters:
  • request – The incoming request object.

  • db_session – The SQLAlchemy Session for database operations.

  • key – The key for the data to retrieve from the session.

  • default – The value to return if the key is not found or session is invalid.

Returns:

The value associated with the key in the session data, or the default value if the session is not found/expired or the key is not present.

get_session_data(db_session: Session, session_id: str) Dict[str, Any] | None[source][source]

Retrieves the full session data dictionary for a given session ID from the database.

This method queries the database for an active (non-expired) session. If found, it returns a dictionary containing all session details, including the raw JSON string from the ‘data’ column.

Parameters:
  • db_session – The SQLAlchemy Session for database operations.

  • session_id – The unique ID of the session to retrieve.

Returns:

A dictionary containing session details (‘id’, ‘user_id’, ‘created_at’, ‘expires_at’, ‘updated_at’, and ‘data’ as a JSON string) if the session is valid and active, otherwise None.

get_user(db_session: Session, session_id: str) Any | None[source][source]

Gets the user ID associated with a session ID from the database, if the session is valid. Relies on get_session_data which emits signals.

Parameters:
  • db_session – The SQLAlchemy Session for database operations.

  • session_id – The ID of the session.

Returns:

The user ID if the session is valid, otherwise None.

is_session_expired(session: Dict[str, Any]) bool[source][source]

Checks if a session dictionary is expired based on its ‘expires_at’ timestamp. # No signals here, as this is an internal helper method. :param session: The session dictionary retrieved from storage (or SQLAlchemy model).

Expected to have an ‘expires_at’ key which is a datetime object.

Returns:

True if the session is expired, False otherwise.

renew_session(db_session: Session, session_id: str) bool[source][source]

Renews the expiration time of an existing session in the database. This typically happens automatically with every request to keep active sessions alive.

Parameters:
  • db_session – The SQLAlchemy database session object.

  • session_id – The unique ID of the session to renew.

Returns:

True if the session was found and successfully renewed, False otherwise.

save_session_data(db_session: Session, session_id: str, data_dict: Dict[str, Any]) bool[source][source]

Saves or updates the entire data payload (as a dictionary) for a specific session in the database.

This method serializes the provided Python dictionary (data_dict) into a JSON string and stores it in the ‘data’ column of the session record. It also renews the session’s expiration time.

Parameters:
  • db_session – The SQLAlchemy Session for database operations.

  • session_id – The unique ID of the session to update.

  • data_dict – The dictionary containing all session data to be stored.

Returns:

True if the session was found and its data was successfully saved, False otherwise.

set_session_data(db_session: Session, session_id: str, key: str, value: Any)[source][source]

Sets a key-value pair within the ‘data’ dictionary of a valid session in the database. This method is used for storing custom session data (like flash messages). Emits ‘session_data_set’ signal on success. Emits ‘session_operation_failed’ signal on failure (session not found/expired). :param db_session: The SQLAlchemy Session for database operations. :param session_id: The ID of the session to update. :param key: The key for the data item within the session’s ‘data’ dictionary. :param value: The value to store. Setting value to None might indicate deletion

depending on how you want to handle it (current implementation stores None).

validate_csrf_token(db_session: Session, session_id: str, csrf_token: str) bool[source][source]

Validates the provided CSRF token against the one stored in the session in the database. Emits ‘csrf_token_validated’ signal on success. Emits ‘csrf_token_validation_failed’ signal on failure. :param db_session: The SQLAlchemy Session for database operations. :param session_id: The ID of the session to validate the token against. :param csrf_token: The token provided in the request (e.g., from form data or header).

Returns:

True if the session is valid and the provided token matches the stored token, False otherwise.

lback.utils.shortcuts module

lback.utils.shortcuts.get_model_form_data(db_session: Session, mapper: Any, obj: Any | None = None) tuple[List[Dict[str, Any]], Dict[str, Dict[str, Any]], List[str]][source][source]

Inspects a SQLAlchemy model’s mapper to determine form fields, relationship fields with choices, and file upload fields. Includes support for Many-to-One, One-to-One, and Many-to-Many relationships. Optionally takes an object to populate current values for change forms. Returns a tuple: (form_fields_data, relationship_fields_data, file_upload_fields).

lback.utils.shortcuts.json_response(data: Any, status_code: int = 200) JSONResponse[source][source]

Returns a JSONResponse.

Parameters:
  • data – The Python data structure to be serialized to JSON.

  • status_code – The HTTP status code for the response (default is 200 OK).

Returns:

A JSONResponse object.

lback.utils.shortcuts.paginate_query(query: Query, page: int = 1, per_page: int = 20) Dict[str, Any][source][source]

Paginates a SQLAlchemy query.

Parameters:
  • query – The SQLAlchemy query object.

  • page – The current page number (1-based index).

  • per_page – The number of items per page.

Returns:

  • objects: List of objects for the current page.

  • total_objects: Total number of objects across all pages.

  • total_pages: Total number of pages.

  • current_page: The current page number.

  • per_page: Items per page.

  • has_prev: Boolean, whether there is a previous page.

  • has_next: Boolean, whether there is a next page.

  • prev_num: Previous page number (None if no previous page).

  • next_num: Next page number (None if no next page).

  • iter_pages: Generator for page numbers to display in pagination control.

Return type:

A dictionary containing pagination data

lback.utils.shortcuts.redirect(to_url: str, status_code: int = 302) RedirectResponse[source][source]

Returns a RedirectResponse.

Parameters:
  • to_url – The URL to redirect to.

  • status_code – The HTTP status code for the redirect (default is 302 Found). Common redirect codes: 301 Moved Permanently, 302 Found, 303 See Other, 307 Temporary Redirect, 308 Permanent Redirect.

Returns:

A RedirectResponse object.

lback.utils.shortcuts.render(request: Request, template_name: str, context: Dict[str, Any] | None = None, status_code: int = 200) Response[source][source]

Renders a template and returns an HTMLResponse.

This is a shortcut function to simplify view functions. It retrieves the template renderer from the request context and uses it to render the specified template with the given context. Also injects common request-related context data into the template.

Parameters:
  • request – The incoming Request object.

  • template_name – The name of the template file (e.g., “admin/login.html”).

  • context – An optional dictionary of context data to pass to the template.

  • status_code – The HTTP status code for the response (default is 200 OK).

Returns:

An HTMLResponse object containing the rendered template content.

Raises:
  • TemplateNotFound – If the specified template does not exist.

  • Exception – For any other errors during rendering.

lback.utils.shortcuts.return_403(request: Request, message: str = 'Forbidden') Response[source][source]

Returns a 403 Forbidden response using the configured error handler if available. Otherwise, returns a basic 403 text response.

Parameters:
  • request – The incoming Request object.

  • message – An optional message (may or may not be used by the error handler).

Returns:

A Response object representing a 403 Forbidden error.

lback.utils.shortcuts.return_404(request: Request, message: str = 'Not Found') Response[source][source]

Returns a 404 Not Found response using the configured error handler if available. Otherwise, returns a basic 404 text response.

Parameters:
  • request – The incoming Request object.

  • message – An optional message (may or may not be used by the error handler).

Returns:

A Response object representing a 404 Not Found error.

lback.utils.shortcuts.return_500(request: Request, message: str = 'Internal Server Error', exception: Exception | None = None) Response[source][source]

Returns a 500 Internal Server Error response using the configured error handler if available. Otherwise, returns a basic 500 text response.

Parameters:
  • request – The incoming Request object.

  • message – An optional message (may or may not be used by the error handler).

  • exception – An optional exception that caused the error, for logging/debugging purposes by the error handler.

Returns:

A Response object representing a 500 Internal Server Error.

lback.utils.static_files module

lback.utils.static_files.find_static_file(config: Any, relative_path: str) str | None[source][source]

Finds the absolute file system path for a static file by searching in STATIC_DIRS.

Parameters:
  • config – The application’s configuration object, expected to have STATIC_DIRS and PROJECT_ROOT attributes.

  • relative_path – The path to the static file relative to the static directories (e.g., ‘css/style.css’).

Returns:

The absolute file system path to the static file, or None if not found.

lback.utils.static_files.static(config: Any, relative_path: str) str[source][source]

Generates the full URL for a static file.

Parameters:
  • config – The application’s configuration object, expected to have a STATIC_URL attribute.

  • relative_path – The path to the static file relative to the static directories (e.g., ‘css/style.css’).

Returns:

The full URL to the static file (e.g., ‘/static/css/style.css’). Returns the relative_path itself if STATIC_URL is not configured.

lback.utils.urls module

lback.utils.urls.path(pattern: str, view_function: Callable[[Request, ...], Any], allowed_methods: List[str] | None = None, name: str | None = None, requires_auth: bool = False) tuple[source][source]

Helper function to define a URL pattern tuple for the router.

Parameters:
  • pattern – The URL pattern string (e.g., ‘/about/’, ‘/users/{user_id:int}/’, ‘/courses/course/{course_slug}/videos/’).

  • view_function – The Python function that handles the request for this pattern.

  • allowed_methods – A list of allowed HTTP methods (e.g., [‘GET’, ‘POST’]). Defaults to [‘GET’] if not provided.

  • name – An optional name for the URL pattern, used for reverse lookups.

  • requires_auth – Boolean indicating if the route requires authentication. Defaults to False.

Returns:

A tuple (pattern, view_function, allowed_methods, name, requires_auth) representing the URL pattern definition, suitable for the router’s urlpatterns list.

lback.utils.user_manager module

class lback.utils.user_manager.UserManager(email_sender: EmailSender, password_validator: PasswordValidator)[source][source]

Bases: object

Service layer for User related business logic. Manages workflows like registration, authentication, and user management. Receives the request-scoped database session per method call and instantiates Repositories with that session. Integrates SignalDispatcher to emit events related to user management.

__init__(email_sender: EmailSender, password_validator: PasswordValidator)[source][source]

Initializes the UserManager. Repositories are instantiated per method call with the request-scoped database session provided to that method.

authenticate_user(session: Session, username: str, password: str, require_email_verified: bool = True) User | None[source][source]

Authenticates user login credentials, with an option to require email verification. Emits ‘user_authentication_started’, ‘user_authenticated’, or ‘user_authentication_failed’ signals.

delete_user(session: Session, user_id: int)[source][source]

Delete a user using the repository. Receives the request-scoped database session. Emits ‘user_deletion_started’, ‘user_pre_delete’, ‘user_deleted’, or ‘user_deletion_failed’ signals.

Parameters:
  • session – The SQLAlchemy Session for database operations.

  • user_id – The ID of the user to delete.

Raises:
  • ValidationError – If user is not found.

  • Exception – For other unexpected errors.

get_all_users(session: Session) List[User][source][source]

Get all users using the repository. Receives the request-scoped database session.

get_user_by_email(session: Session, email: str) User | None[source][source]

Get a user by their email address using the repository. Receives the request-scoped database session.

get_user_by_id(session: Session, user_id: int) User | None[source][source]

Get a user by their ID using the repository. Receives the request-scoped database session.

get_user_by_username(session: Session, username: str) User | None[source][source]

Get a user by their username using the repository. Receives the request-scoped database session.

register_user(session: Session, username: str, email: str, plain_password: str) User | None[source][source]

Registers a new user with validation, password complexity checks, and initiates email verification. Assigns the user to the ‘basic_user’ group. Emits ‘user_registration_started’, ‘user_pre_register’, ‘user_registered’, ‘user_email_verification_initiated’, or ‘user_registration_failed’ signals.

reset_password(session: Session, token: str, new_password: str) bool[source][source]

Resets a user’s password using a password reset token. Verifies the token, updates the password, and clears the token.

reset_password_request(session: Session, email: str, reset_url_path: str | None = None) bool[source][source]

Initiates a password reset request for a user. Generates a reset token, stores it on the user, and prepares for email sending.

Parameters:
  • session – The SQLAlchemy Session.

  • email – The email of the user requesting reset.

  • reset_url_path – The base URL path to use for the reset link (e.g., “/reset-password-confirm/”). If None, it defaults to the API path.

search_users(session: Session, **criteria: Any) List[User][source][source]

Search for users based on provided criteria using the repository or session. Receives the request-scoped database session.

Parameters:
  • session – The SQLAlchemy Session for database operations.

  • **criteria – Keyword arguments for search criteria (e.g., username=”test”).

Returns:

A list of User objects matching the criteria.

update_user(session: Session, user_id: int, data: Dict[str, Any]) User | None[source][source]

Update an existing user’s data using the repository. Handles password hashing, and can update email verification status. Note: Group membership changes should be handled explicitly, not via ‘data’ dict.

verify_user_email(session: Session, token: str) User | None[source][source]

Verifies a user’s email using the provided token. Emits ‘user_email_verification_successful’ or ‘user_email_verification_failed’ signals.

lback.utils.validation module

class lback.utils.validation.PasswordValidator[source][source]

Bases: object

Provides methods to validate password complexity.

validate(password: str)[source][source]

Validates the complexity of a given password.

Parameters:

password – The plain text password to validate.

Raises:

ValidationError – If the password does not meet complexity requirements.

exception lback.utils.validation.ValidationError(message='Validation error occurred')[source][source]

Bases: Exception

Custom exception for validation errors.

lback.utils.validation.validate_json(request, required_fields: dict, optional_fields: dict = None)[source][source]

Validate JSON data in the request.

Parameters:
  • request – The request object containing parsed_body.

  • required_fields (dict) – A dictionary of required fields and their expected types.

  • optional_fields (dict) – A dictionary of optional fields and their expected types.

Returns:

The validated data.

Return type:

dict

Raises:

ValidationError – If validation fails.