wy_qcos.api.posiq.routes_jsonrpc package
Subpackages
Submodules
wy_qcos.api.posiq.routes_jsonrpc.auth module
- async wy_qcos.api.posiq.routes_jsonrpc.auth.login(request, body, user_manager=Depends(get_user_manager), users_repo=Depends(get_repository(UserRepository)))
Authenticate user and return JWT token.
This endpoint validates user credentials and generates a JWT access token for authenticated users. The token is used for subsequent API requests.
- 参数:
request (fastapi.Request) -- HTTP request object
body (LoginRequest) -- Login request containing username and password
user_manager (UserManager) -- UserManager dependency for user operations
users_repo (UserRepository) -- User repository dependency for database operations
- 返回:
LoginResponse containing JWT access token and expiration info
- 抛出:
UnauthorizedError -- If credentials are invalid or user is disabled
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.auth.logout(request, body=None, auth_data=Depends(auth), user_manager=Depends(get_user_manager))
Logout user and invalidate token.
This endpoint adds the current token to the blacklist, making it invalid for subsequent requests. The client should also remove the token from local storage.
- 参数:
request (fastapi.Request) -- HTTP request object
body (LogoutRequest | None) -- Optional logout request
auth_data (dict | None) -- Authentication data from current user
user_manager (UserManager) -- UserManager instance for blacklist operations
- 返回:
LogoutResponse with confirmation message
- 返回类型:
- async wy_qcos.api.posiq.routes_jsonrpc.auth.refresh_token(request, body, user_manager=Depends(get_user_manager))
Refresh JWT token using refresh_token.
This endpoint follows the standard JWT refresh token pattern: - Client provides a valid refresh_token - Server validates the refresh_token - Server returns new access_token and refresh_token - Old refresh_token is optionally invalidated
- 参数:
request (fastapi.Request) -- HTTP request object
body (TokenRefreshRequest) -- Token refresh request containing refresh_token
user_manager (UserManager) -- UserManager dependency for user operations
- 返回:
TokenRefreshResponse with new JWT access and refresh tokens
- 抛出:
UnauthorizedError -- If refresh token is invalid or expired
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.auth.me(auth_data=Depends(auth), user_manager=Depends(get_user_manager))
Get current authenticated user information.
This endpoint returns the profile information of the currently authenticated user based on their JWT token.
- 参数:
auth_data (dict | None) -- Authentication data from current user
user_manager (UserManager) -- UserManager dependency for user operations
- 返回:
GetUserResponse with current user's profile information
- 抛出:
UnauthorizedError -- If user is not authenticated
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.device module
- wy_qcos.api.posiq.routes_jsonrpc.device.get_devices(body=None, auth_data=Depends(auth))
Get device dict request.
- 参数:
body (schemas.GetDevicesRequest) -- devices request
auth_data (dict | None) -- auth data
- 返回:
Get devices response
- 返回类型:
dict[str, GetDeviceResponse]
- wy_qcos.api.posiq.routes_jsonrpc.device.get_device(body, auth_data=Depends(auth))
Get device info request.
- 参数:
body (schemas.GetDeviceRequest) -- device name
auth_data (dict | None) -- auth data
- 返回:
Get device info response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.device.calibrate_device(body, auth_data=Depends(auth))
Calibrate device.
- 参数:
body (schemas.CalibrateDeviceRequest) -- CalibrateDeviceRequest body
auth_data (dict | None) -- auth data
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.device.get_calibrate_results(body, auth_data=Depends(auth))
Get Calibrate result.
- 参数:
body (schemas.GetCalibrateResultRequest) -- GetCalibrateResultRequest body
auth_data (dict | None) -- auth data
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.device.set_device_options(body, auth_data=Depends(auth))
Set device Options request.
- 参数:
body (schemas.SetDeviceOptionsRequest) -- SetDeviceOptionsRequest body
auth_data (dict | None) -- auth data
- 返回:
Set device Options response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.device.get_device_options(body, auth_data=Depends(auth))
Get device Options request.
- 参数:
body (schemas.GetDeviceOptionsRequest) -- GetDeviceOptionsRequest body
auth_data (dict | None) -- auth data
- 返回:
Set device Options response
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.driver module
- wy_qcos.api.posiq.routes_jsonrpc.driver.get_drivers(body=None, auth_data=Depends(auth))
Get driver dict request.
- 参数:
body (schemas.GetDriversRequest) -- message
auth_data (dict | None) -- auth data
- 返回:
Get drivers response
- 返回类型:
dict[str, GetDriverResponse]
- wy_qcos.api.posiq.routes_jsonrpc.driver.get_driver(body, auth_data=Depends(auth))
Get driver info request.
- 参数:
body (schemas.GetDriverRequest) -- driver_name
auth_data (dict | None) -- auth data
- 返回:
Get driver info response
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.errors module
JSON-RPC/RestfulAPI error-code mappings.
Error codes mappings and descriptions: .. code-block:: text
err_code
http_code
Description
Examples / Scenarios
0 -400 -401 -403 -404 -409
-500 -501 -503
2XX 400 401 403 404 409
500 501 503
Accepted / OK Bad Request Unauthorized Forbidden Not Found Conflict
Internal Server Error Not Implemented Service Unavailable
success invalid params / request unauthorized login / token unauthorized operations resource not found create duplicated resource name resource deps not met bug / exception not implemented service offline
- class wy_qcos.api.posiq.routes_jsonrpc.errors.JsonRpcBaseError(*args, **kwargs)
基类:
BaseErrorJsonRpc Base Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- pydantic model DataModel
基类:
BaseModelData Model.
Show JSON schema
{ "title": "DataModel", "description": "Data Model.", "type": "object", "properties": { "details": { "title": "Details", "type": "string" } }, "required": [ "details" ] }
- Fields:
details (str)
- field details: str [Required]
- class wy_qcos.api.posiq.routes_jsonrpc.errors.BadRequestError(*args, **kwargs)
-
Bad Request Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -400
- MESSAGE: str | None = 'Bad Request'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.UnauthorizedError(*args, **kwargs)
-
Unauthorized Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -401
- MESSAGE: str | None = 'Unauthorized'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.ForbiddenError(*args, **kwargs)
-
Forbidden Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -403
- MESSAGE: str | None = 'Forbidden'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.NotFoundError(*args, **kwargs)
-
Not Found Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -404
- MESSAGE: str | None = 'Not Found'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.ConflictError(*args, **kwargs)
-
Conflict Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -409
- MESSAGE: str | None = 'Conflict'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.InternalServerError(*args, **kwargs)
-
Internal Server Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -500
- MESSAGE: str | None = 'Internal Server Error'
- class wy_qcos.api.posiq.routes_jsonrpc.errors.NotImplementedError(*args, **kwargs)
-
Not Implemented Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- CODE: int | None = -501
- MESSAGE: str | None = 'Not Implemented'
-
Service Unavailable Error.
- 参数:
args (Any)
kwargs (Any)
- 返回类型:
Any
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_errors(err_cls, module_name, func_name, results, param_name, code)
Handle errors.
- 参数:
err_cls -- error class
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param
code -- error code
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_bad_requests(module_name, func_name, results, param_name=None, code=None)
Handle bad_requests error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_unauthorized(module_name, func_name, results, param_name=None, code=None)
Handle unauthorized error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_forbidden(module_name, func_name, results, param_name=None, code=None)
Handle forbidden error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_not_found(module_name, func_name, results, param_name=None, code=None)
Handle forbidden error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_conflict(module_name, func_name, results, param_name=None, code=None)
Handle conflict error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_internal_server(module_name, func_name, results, param_name=None, code=None)
Handle internal server error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
- wy_qcos.api.posiq.routes_jsonrpc.errors.handle_error_not_implemented(module_name, func_name, results, param_name=None, code=None)
Handle not implemented error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
Handle service unavailable error.
- 参数:
module_name -- module name
func_name -- function name
results -- results for jsonrpc
param_name -- name of the param (Default value = None)
code -- error code (Default value = None)
wy_qcos.api.posiq.routes_jsonrpc.job module
- wy_qcos.api.posiq.routes_jsonrpc.job.submit_job(body, auth_data=Depends(auth))
Submit job.
- 参数:
body (schemas.SubmitJobRequest) -- job info
auth_data (dict | None) -- auth data
- 返回:
job info
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.job.get_job_status(body, auth_data=Depends(auth))
Get job status.
- 参数:
body (schemas.GetJobStatusRequest) -- job_id: job ID
auth_data (dict | None) -- auth data
- 返回:
job status
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.job.get_job_results(body, auth_data=Depends(auth))
Get job results.
- 参数:
body (schemas.GetJobResultsRequest) -- job_id: job ID
auth_data (dict | None) -- auth data
- 返回:
job results
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.job.get_jobs(body=None, auth_data=Depends(auth))
Get job list.
- 参数:
body (schemas.GetJobsRequest) -- job_id: job ID
auth_data (dict | None) -- auth data
- 返回:
job list
- 返回类型:
list[GetJobStatusResponse]
- wy_qcos.api.posiq.routes_jsonrpc.job.cancel_jobs(body, auth_data=Depends(auth))
Cancel job.
- 参数:
body (schemas.CancelJobsRequest) -- job_ids: job IDs
auth_data (dict | None) -- auth data
- 返回:
cancelled jobs info
- 返回类型:
list[CancelJobsResponse]
- wy_qcos.api.posiq.routes_jsonrpc.job.delete_jobs(body, auth_data=Depends(auth))
Delete job.
- 参数:
body (schemas.DeleteJobsRequest) -- job_ids: job IDs
auth_data (dict | None) -- auth data
- 返回:
deleted jobs info
- 返回类型:
list[DeleteJobsResponse]
- wy_qcos.api.posiq.routes_jsonrpc.job.set_job_results(body, auth_data=Depends(auth))
Set job results for existing job.
- 参数:
body (schemas.SetJobResultsRequest) -- job_id: job ID
auth_data (dict | None) -- auth data
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.job.update_job(body, auth_data=Depends(auth))
Update job.
- 参数:
body (schemas.UpdateJobsRequest) -- job info
auth_data (dict | None) -- auth data
- 返回:
update job param
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.job.merge_results(response_info, parameters, results=None)
Merge results.
- 参数:
response_info -- response info
parameters -- parameters from prefect
results -- results from prefect (Default value = None)
- 返回:
new response info
wy_qcos.api.posiq.routes_jsonrpc.metrics module
- wy_qcos.api.posiq.routes_jsonrpc.metrics.get_system_health(body=None, auth_data=Depends(auth))
Get system health status.
- 参数:
body (GetSystemHealthRequest | None) -- Get system health request
auth_data (dict | None) -- Authentication data
- 返回:
System health response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.metrics.get_api_stats(body=None, auth_data=Depends(auth))
Get API access statistics.
- 参数:
body (GetApiStatsRequest | None) -- Get API statistics request
auth_data (dict | None) -- Authentication data
- 返回:
API statistics response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.metrics.get_job_stats(body=None, auth_data=Depends(auth))
Get job statistics.
- 参数:
body (GetJobStatsRequest | None) -- Get job statistics request
auth_data (dict | None) -- Authentication data
- 返回:
Job statistics response
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.routes module
wy_qcos.api.posiq.routes_jsonrpc.system module
- wy_qcos.api.posiq.routes_jsonrpc.system.ping(body, auth_data=Depends(auth))
Ping-pong to verify the availability of the system.
- 参数:
body (schemas.PingRequest) -- message
auth_data (dict | None) -- auth data
- 返回:
pong response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.system.system_info(body=None, auth_data=Depends(auth))
Get system info.
- 参数:
body (schemas.SystemInfoRequest) -- System Info Request
auth_data (dict | None) -- auth data
- 返回:
system info response
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.transpiler module
- wy_qcos.api.posiq.routes_jsonrpc.transpiler.get_transpilers(body=None, auth_data=Depends(auth))
Get transpiler list request.
- 参数:
body (schemas.GetTranspilersRequest) -- message
auth_data (dict | None) -- auth data
- 返回:
Get transpilers response
- 返回类型:
dict[str, GetTranspilerResponse]
- wy_qcos.api.posiq.routes_jsonrpc.transpiler.get_transpiler(body, auth_data=Depends(auth))
Get transpiler info request.
- 参数:
body (schemas.GetTranspilerRequest) -- driver_name
auth_data (dict | None) -- auth data
- 返回:
Get transpiler info response
- 返回类型:
wy_qcos.api.posiq.routes_jsonrpc.user module
- wy_qcos.api.posiq.routes_jsonrpc.user.get_user_manager(request)
Get user manager.
- 参数:
request (fastapi.Request) -- request object
- 返回:
user manager object
- wy_qcos.api.posiq.routes_jsonrpc.user.get_user_mgmt(body=None, auth_data=Depends(auth))
Get user management status.
- 参数:
body (GetUserMgmtRequest | None) -- request body
auth_data (dict | None) -- auth data
- 返回:
User management status response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.set_user_mgmt(body, auth_data=Depends(auth))
Set user management authentication mode.
- 参数:
body (SetUserMgmtRequest) -- request body with auth_mode
auth_data (dict | None) -- auth data
- 返回:
SetUserMgmtResponse with updated auth_mode
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.create_user(body, request, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)), roles_repo=Depends(get_repository(RoleRepository)))
Create a new user.
- 参数:
body (CreateUserRequest) -- user creation request
request (fastapi.Request) -- request object
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Create user response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.get_user(body, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Get user information by ID.
- 参数:
body (GetUserRequest) -- get user request (contains user_id)
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
Get user response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.get_users(body=None, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Get users with optional filtering.
- 参数:
body (GetUsersRequest | None) -- get users request with optional filter dict
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
Dictionary of users keyed by user_name
- 返回类型:
dict[str, GetUserResponse]
- Filter example:
{"user_name": "admin"} - filter by user_name
- wy_qcos.api.posiq.routes_jsonrpc.user.update_user(body, request, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)), roles_repo=Depends(get_repository(RoleRepository)))
Update user information by ID.
- 参数:
body (UpdateUserRequest) -- user update request (contains user_id)
request (fastapi.Request) -- request object
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Update user response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.delete_user(body, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Delete user by ID.
- 参数:
body (DeleteUserRequest) -- delete user request (contains user_id and optional force flag)
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
Delete user response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.create_role(body, request, auth_data=Depends(auth), roles_repo=Depends(get_repository(RoleRepository)))
Create a new role.
- 参数:
body (CreateRoleRequest) -- role creation request
request (fastapi.Request) -- request object
auth_data (dict | None) -- auth data
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Create role response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.get_role(body, auth_data=Depends(auth), roles_repo=Depends(get_repository(RoleRepository)))
Get role information by ID.
- 参数:
body (GetRoleRequest) -- get role request (contains role_id)
auth_data (dict | None) -- auth data
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Get role response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.get_roles(body=None, auth_data=Depends(auth), roles_repo=Depends(get_repository(RoleRepository)))
Get roles with optional filtering.
- 参数:
body (GetRolesRequest | None) -- get roles request with optional filter dict
auth_data (dict | None) -- auth data
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Dictionary of roles keyed by role name
- 返回类型:
dict[str, GetRoleResponse]
- Filter example:
{"role_name": "admin"} - filter by role_name
- wy_qcos.api.posiq.routes_jsonrpc.user.update_role(body, request, auth_data=Depends(auth), roles_repo=Depends(get_repository(RoleRepository)))
Update role information by ID.
- 参数:
body (UpdateRoleRequest) -- role update request (contains role_id)
request (fastapi.Request) -- request object
auth_data (dict | None) -- auth data
roles_repo (RoleRepository) -- Role repository dependency
- 返回:
Update role response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.delete_role(body, request, auth_data=Depends(auth), roles_repo=Depends(get_repository(RoleRepository)), users_repo=Depends(get_repository(UserRepository)))
Delete role by ID.
- 参数:
body (DeleteRoleRequest) -- delete role request (contains role_id)
request (fastapi.Request) -- request object
auth_data (dict | None) -- auth data
roles_repo (RoleRepository) -- Role repository dependency
users_repo (UserRepository) -- User repository dependency
- 返回:
Delete role response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.change_password(body, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Change user password by ID.
- 参数:
body (ChangePasswordRequest) -- password change request (contains user_id)
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
Change password response
- 返回类型:
- wy_qcos.api.posiq.routes_jsonrpc.user.get_login_logs(body=None, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Get login logs by user ID or user_name.
- 参数:
body (GetLoginLogsRequest | None) -- get login logs request (contains user_id, user_name, limit, offset)
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
List of login logs in descending order by login_time
- 返回类型:
list[LoginLogResponse]
备注
user_id and user_name are mutually exclusive. Only one can be provided. If both are None, all login logs will be returned.
- wy_qcos.api.posiq.routes_jsonrpc.user.clear_login_logs(body=None, auth_data=Depends(auth), users_repo=Depends(get_repository(UserRepository)))
Clear login logs (all or for a specific user).
- 参数:
body (ClearLoginLogsRequest | None) -- clear login logs request (contains user_id, user_name)
auth_data (dict | None) -- auth data
users_repo (UserRepository) -- User repository dependency
- 返回:
Dictionary with count of deleted logs
- 返回类型:
dict
备注
user_id and user_name are mutually exclusive. Only one can be provided. If both are None, all login logs will be cleared.
- wy_qcos.api.posiq.routes_jsonrpc.user.get_user_response(user)
Get user response.
- 参数:
user (schemas.User) -- User model instance
- 返回:
Formatted user response
- 返回类型:
schemas.GetUserResponse
- wy_qcos.api.posiq.routes_jsonrpc.user.get_role_response(role)
Get role response.
- 参数:
role -- role info
- 返回:
role response
- 返回类型:
dict
wy_qcos.api.posiq.routes_jsonrpc.version module
- wy_qcos.api.posiq.routes_jsonrpc.version.version(body)
Get server version.
- 参数:
body (GetVersionRequest) -- schemas.GetVersionRequest: (Default value = None)
- 返回类型: