wy_qcos.db.repositories package

Submodules

wy_qcos.db.repositories.base module

exception wy_qcos.db.repositories.base.ControllerDatabaseError(message)

基类：Exception

参数:

message (str)

class wy_qcos.db.repositories.base.BaseRepository(db_session)

基类：object

参数:

db_session (Session)

create(model_class, **kwargs)

Create a record in table.

参数:

  • model_class (type)

  • kwargs (Any)

get_by_attr(model_class, attr_name, attr_value, child_attr_name=None, unique=True)

Get a record from table by attribute.

参数:

  • model_class (type)

  • attr_name (str)

  • child_attr_name (str | None)

  • unique (bool | None)

get_by_uuid(model_class, uuid, child_attr_name=None)

Get a record from table by UUID string.

参数:

  • model_class (type)

  • uuid (str)

  • child_attr_name (str | None)

get_all(model_class, child_attr_name=None)

Get all records.

参数:

  • model_class (type)

  • child_attr_name (str | None)

update(model_class, uuid, **kwargs)

Update a record with UUID string using args.

参数:

  • model_class (type)

  • uuid (str)

  • kwargs (Any)

delete_by_uuid(model_class, uuid)

Delete a record from table by UUID string.

参数:

  • model_class (type)

  • uuid (str)

delete_by_attr(model_class, attr_name, attr_value)

Delete a record from table by attribute.

参数:

  • model_class (type)

  • attr_name (str)

rollback()
返回类型:

None

commit()
返回类型:

None

wy_qcos.db.repositories.project module

class wy_qcos.db.repositories.project.ProjectRepository(db_session)

基类：BaseRepository

Database operation function library related to Projects.

参数:

db_session (Session)

get_project_by_id(project_id)

Get project by ID.

参数:

project_id (str) -- project ID

返回:

tuple of (success, error, project)

返回类型:

tuple[bool, str | None, Project | None]

get_project_by_name(name)

Get project by name.

参数:

name (str) -- project name

返回:

tuple of (success, error, project)

返回类型:

tuple[bool, str | None, Project | None]

get_projects()

Get all projects.

返回:

tuple of (success, error, projects)

返回类型:

tuple[bool, str | None, list[Project] | None]

create_project(project_id, name)

Create a project.

参数:

  • project_id (str) -- project ID (UUID string)

  • name (str) -- project name

返回:

tuple of (success, error, project)

返回类型:

tuple[bool, str | None, Project | None]

project_exists(project_id)

Check if project exists.

参数:

project_id (str) -- project ID

返回:

True if project exists, False otherwise

返回类型:

bool

wy_qcos.db.repositories.role module

class wy_qcos.db.repositories.role.RoleRepository(db_session)

基类：BaseRepository

Database operation function library related to Roles.

参数:

db_session (Session)

create_role(role_create)

Create a new role.

参数:

role_create (CreateRoleRequest)

get_role_by_id(role_id)

Get a role by UUID string ID.

参数:

role_id (str)

get_role_by_name(role_name)

Get a role by name.

参数:

role_name (str)

get_roles()

Get all roles.

update_role(role_id, role_update)

Update a role.

参数:
delete_role_by_id(role_id)

Delete a role by UUID string.

参数:

role_id (str)

wy_qcos.db.repositories.user module

class wy_qcos.db.repositories.user.UserRepository(db_session)

基类：BaseRepository

Database operation function library related to Users.

参数:

db_session (Session)

static hash_password(password)

Hash a password.

参数:

password (str)

返回类型:

str

static verify_password(plain_password, hashed_password)

Verify a password against a hash.

参数:

  • plain_password (str)

  • hashed_password (str)

返回类型:

bool

create_user(user_create)

Create a new user.

参数:

user_create (CreateUserRequest)

get_user_by_username(user_name)

Get a user by username.

参数:

user_name (str)

get_user_by_id(user_id)

Get a user by UUID string ID.

参数:

user_id (str)

get_users()

Get all users.

update_user(user_id, user_update)

Update a user.

参数:
delete_user_by_id(user_id)

Delete a user by UUID string.

This method first deletes all user-role associations before deleting the user to avoid foreign key constraint violations.

参数:

user_id (str)

delete_user_by_username(user_name)

Delete a user by username.

This method first deletes all user-role associations before deleting the user to avoid foreign key constraint violations.

参数:

user_name (str)

create_login_log(user_name, ip_address, success, failure_reason=None, user_agent=None, project_id=None)

Create a login log entry with auto cleanup when logs exceeded.

参数:

  • user_name (str)

  • ip_address (str)

  • success (bool)

  • failure_reason (str | None)

  • user_agent (str | None)

  • project_id (str | None)

get_login_logs(user_id=None, start_time=None, end_time=None, limit=100, offset=0)

Get login logs with optional filters.

参数:

  • user_id (str | None)

  • start_time (datetime | None)

  • end_time (datetime | None)

  • limit (int)

  • offset (int)

delete_login_logs(user_id=None, user_name=None)

Delete login logs (all or for a specific user).

参数:

  • user_id (str | None) -- User ID (UUID) to delete logs for (optional)

  • user_name (str | None) -- User name to delete logs for (optional)

返回:

Tuple (success, error, count) where count is number of deleted logs

add_to_blacklist(token_jti, expires_at)

Add a token to the blacklist.

参数:

  • token_jti (str)

  • expires_at (datetime)

is_blacklisted(token_jti)

Check if a token is blacklisted.

参数:

token_jti (str)

返回类型:

bool

cleanup_blacklist()

Remove expired entries from the blacklist.

assign_role(user_id, role_name)

Assign a role to a user.

参数:

  • user_id (str) -- User ID

  • role_name (str) -- Role name to assign

返回:

Tuple of (success, error_message)

返回类型:

tuple[bool, str | None]

remove_role(user_id, role_name)

Remove a role from a user.

参数:

  • user_id (str) -- User ID

  • role_name (str) -- Role name to remove

返回:

Tuple of (success, error_message)

返回类型:

tuple[bool, str | None]

revoke_role(user_id, role_name)

Revoke a role from a user (alias for remove_role).

参数:

  • user_id (str) -- User ID

  • role_name (str) -- Role name to revoke

返回:

Tuple of (success, error_message)

返回类型:

tuple[bool, str | None]

update_user_roles(user_id, role_names)

Update all roles for a user (replace existing roles).

参数:

  • user_id (str) -- User ID

  • role_names (list[str]) -- List of role names to assign

返回:

Tuple of (success, error_message)

返回类型:

tuple[bool, str | None]

get_user_roles(user_id)

Get all roles for a user.

参数:

user_id (str) -- User ID

返回:

Tuple of (success, error_message, role_names_list)

返回类型:

tuple[bool, str | None, list[str]]

Module contents