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, filters=None)

Get a record from table by UUID string with optional filters.

参数:

  • model_class (type) -- The model class to query

  • uuid (str) -- The UUID string or object

  • child_attr_name (str | None) -- Optional child attribute name for eager loading

  • filters (dict | None) -- Optional dictionary with additional filter conditions Example: {'project_id': 'xxx', 'user_id': 'yyy'} Combines with id filter using AND logic

返回:

(success, error, record)

返回类型:

Tuple[bool, Exception|None, record]

get_all(model_class, child_attr_name=None, filters=None)

Get all records with optional filtering.

参数:

  • model_class (type) -- The model class to query

  • child_attr_name (str | None) -- Optional child attribute name for eager loading

  • filters (dict | None) -- Dictionary with filter conditions. Each key is a column attribute name, value is the filter value. Example: {'project_id': 'xxx', 'user_id': 'yyy', 'status': 'ACTIVE'} Multiple filters are combined with AND logic

返回:

(success, error, records)

返回类型:

Tuple[bool, Exception|None, list]

count(model_class)

Get count of all records in table.

参数:

model_class (type) -- The model class to count

返回:

Count of records, returns 0 on error

返回类型:

int

count_by_attr(model_class, attr_name, attr_value)

Get count of records by attribute value.

参数:

  • model_class (type) -- The model class to count

  • attr_name (str) -- The attribute name to filter by

  • attr_value -- The attribute value to match

返回:

Count of matching records, returns 0 on error

返回类型:

int

count_with_filters(model_class, filters=None)

Get count of records with optional filtering.

参数:

  • model_class (type) -- The model class to count

  • filters (dict | None) --

    Dictionary with filter conditions. Each key is a column attribute name, value is the filter value. Multiple filters are combined with AND logic. Example:

    {
    "project_id": "xxx",
    "user_id": "yyy",
    "job_status": "COMPLETED",
}

返回:

Count of matching records, returns 0 on error

返回类型:

int

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

flush()

Flush pending changes to database without committing.

返回类型:

None

refresh(obj)

Refresh an ORM object to get latest committed data.

返回类型:

None

wy_qcos.db.repositories.job module

class wy_qcos.db.repositories.job.JobRepository(db_session)

基类：BaseRepository

Database operation function library related to Job.

参数:

db_session (Session)

create_job(job_create, auto_commit=True)

Create a new job.

参数:

  • job_create (SubmitJobRequest) -- Job creation request data

  • auto_commit (bool) -- If True, automatically commit the transaction. If False, only add to session (requires manual commit). In both cases, flush is called to populate object IDs.

返回:

(success, error, job_record)

返回类型:

Tuple[bool, Exception|None, Job|None]

get_job_by_uuid(job_id, filters={})

Get job by uuid.

参数:

  • job_id (UUID) -- UUID

  • filters -- db filters

返回:

job records by uuid with filters

get_jobs(filters=None)

Get jobs with optional filtering.

参数:

filters (dict | None) -- Dictionary with filter conditions. Supported keys are 'id', 'uuid', 'project_id', 'user_id', 'job_status', 'code_type', 'backend', 'job_name', 'is_callback_success'.

Example:

# No filter - get all jobs
success, error, jobs = self.get_jobs()

# Single filter
success, error, jobs = self.get_jobs(filters={"project_id": "xxx"})

# Multiple filters (AND condition)
success, error, jobs = self.get_jobs(
    filters={
        "project_id": "xxx",
        "user_id": "yyy",
        "job_status": "COMPLETED",
    }
)
返回:

Tuple[bool, Exception|None, list[Job]|None]

参数:

filters (dict | None)

get_jobs_count(filters=None)

Get count of jobs with optional filtering.

参数:

filters (dict | None) -- Dictionary with filter conditions. Supported keys are 'project_id', 'user_id', 'job_status', 'code_type', 'backend', 'job_name', 'is_callback_success'.

返回类型:

int

Example:

# Get total jobs count
count = self.count_jobs()

# Count jobs with specific status
count = self.count_jobs(filters={"job_status": "COMPLETED"})

# Multiple filters (AND condition)
count = self.count_jobs(
    filters={"project_id": "xxx", "job_status": "QUEUED"}
)
返回:

Count of matching jobs, returns 0 on error

参数:

filters (dict | None)

返回类型:

int

update_job_results(job_id, job_update)

Update a job.

Only updates fields that are explicitly set (non-None values are included). None values are excluded to prevent overwriting existing data.

参数:

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, description=None)

Create a project.

参数:

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

  • name (str) -- project name

  • description (str | None) -- project description (optional)

返回:

tuple of (success, error, project)

返回类型:

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

update_project(project_id, name=None, description=None)

Update a project.

参数:

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

  • name (str | None) -- new project name (optional)

  • description (str | None) -- new project description (optional)

返回:

tuple of (success, error, project)

返回类型:

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

delete_project(project_id)

Delete a project.

参数:

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

返回:

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(filters=None)

Get all roles with optional filtering.

参数:

filters (dict | None) -- Dictionary with filter conditions (e.g., {'role_name': 'admin'})

返回:

Tuple of (success, error, 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(filters=None)

Get all users with optional filtering.

参数:

filters (dict | None) -- Dictionary with filter conditions (e.g., {'user_name': 'admin', 'is_enabled': True})

返回:

Tuple of (success, error, 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 associated job records, then 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 associated job records, then 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) -- Filter by user ID (optional)

  • start_time (datetime | None) -- Filter logs after this time (optional)

  • end_time (datetime | None) -- Filter logs before this time (optional)

  • limit (int) -- Maximum number of logs to return. Use -1 to get all logs without limit

  • offset (int) -- Number of logs to skip

返回:

Tuple of (success, error, logs)

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