interfaces - base classes

class Cache[source]

Bases: ABC

Data key-value store.

namespace: Namespace
abstract async get(id: NSKey) → Any[source]
abstract async m_get(id: Collection[NSKey]) → dict[kaiju_tools.types.NSKey, Any][source]
abstract async exists(id: NSKey) → bool[source]
abstract async m_exists(id: Collection[NSKey]) → frozenset[kaiju_tools.types.NSKey][source]
abstract async set(id: NSKey, data, ttl: int = None) → None[source]
abstract async m_set(data: dict[kaiju_tools.types.NSKey, Any], ttl: int = None) → None[source]
abstract async delete(id: NSKey) → None[source]
abstract async m_delete(id: Collection[NSKey]) → None[source]
class DataStore[source]

Bases: ABC

Data row-column store.

abstract async get(id: Hashable, columns: Collection[str] | Literal['*'] | None = '*', _connection=None)[source]
abstract async m_get(id: Collection[Hashable], columns: Collection[str] | Literal['*'] | None = '*', _connection=None)[source]
abstract async exists(id: Hashable, _connection=None)[source]
abstract async m_exists(id: Collection[Hashable], _connection=None)[source]
abstract async delete(id: Hashable, columns: Collection[str] | Literal['*'] | None = None, _connection=None)[source]
abstract async m_delete(id: Collection[Hashable] = None, conditions: dict = None, columns: Collection[str] | Literal['*'] | None = None, _connection=None)[source]
abstract async create(data: dict, columns: Collection[str] | Literal['*'] | None = '*', _connection=None, on_conflict: str = None, on_conflict_keys: Collection = None, on_conflict_values=None)[source]
abstract async m_create(data: Collection, columns: Collection[str] | Literal['*'] | None = '*', _connection=None, on_conflict: str = None, on_conflict_keys: Collection = None, on_conflict_values: dict = None)[source]
abstract async update(id: Hashable, data, columns: Collection[str] | Literal['*'] | None = '*', _connection=None)[source]
abstract async m_update(id: Collection[Hashable], data, conditions: dict = None, columns: Collection[str] | Literal['*'] | None = '*', _connection=None)[source]
abstract async iter(conditions: dict = None, sort=None, offset: int = 0, limit: int = 10, columns: Collection[str] | Literal['*'] | None = '*') → AsyncGenerator[list, None][source]
class UserInterface[source]

Bases: Generic[_User], ABC

User login interface.

abstract async auth(username: str, password: str) → _User | None[source]

User login and password check.

Must return None if no user or token is invalid.

abstract async register(username: str, email: str, password: str, settings: dict = None) → _User[source]

Register a new user.

abstract async change_password(username: str, password: str, new_password: str)[source]

Change user password.

abstract async update_profile(id: UUID, settings: dict)[source]

Update current user profile.

abstract async get_user_and_permissions(id: UUID)[source]

Get user and user permissions info.

class TokenInterface[source]

Bases: Generic[_User], ABC

Auth token interface.

class TokenClaims[source]

Bases: TypedDict

JWT token claims data.

id: UUID
permissions: Collection[str]
class TokenInfo[source]

Bases: TypedDict

JWT methods output.

access: str
refresh: str
abstract async auth(token: str, /) → TokenClaims | None[source]

Verify an auth token and return token user data.

Must return None if no user or token is invalid.

abstract async get(claims: TokenClaims, /) → TokenInfo[source]

Generate a token pair (access / refresh tokens).

abstract async refresh(token: str, /) → TokenInfo | None[source]

Generate a token pair (access / refresh tokens).

AbstractRPCCompatible

alias of PublicInterface

class PublicInterface[source]

Bases: ABC

Class with an RPC interface.

DEFAULT_PERMISSION = '*'
class PermissionKeys[source]

Bases: object

Permission scopes.

GLOBAL_SYSTEM_PERMISSION = 0
GLOBAL_USER_PERMISSION = 100
GLOBAL_GUEST_PERMISSION = 1000
app: App
get_session()[source]

Get current user session.

get_request_context() → RequestContext | None[source]

Get current user request context.

get_user_id()[source]

Return current session user id.

has_permission(permission: str) → bool[source]

Check if a user session has a particular permission.

system_user() → bool[source]

Check if user session has the system scope.

property routes: dict

List RPC routes.

property permissions: dict

List RPC routes permissions.

property validators: dict

List of RPC routes validation schemas.

class SessionInterface[source]

Bases: Generic[_Session], ABC

Session management interface.

abstract get_new_session(data: dict, *, user_agent: str | bytes = None) → _Session[source]
abstract async load_session(session_id: str, /, *, user_agent: str = None) → _Session | None[source]
abstract async session_exists(session_id: str, /) → bool[source]
abstract async save_session(session: _Session, /) → None[source]
abstract async delete_session(session: _Session, /) → None[source]
class AuthenticationInterface[source]

Bases: Generic[_Session], ABC

Authentication from auth strings / headers / tokens.

abstract async header_auth(auth_string: str, /) → _Session | None[source]
abstract async basic_auth(auth_string: str, /) → _Session[source]
abstract async password_auth(session: _Session, username: str, password: str) → _Session[source]
abstract async token_auth(token: str, /) → _Session[source]
class Locks[source]

Bases: ABC

Shared (between apps) locks management.

LockId = kaiju_tools.interfaces.LockId
abstract async acquire(id: NSKey, identifier: LockId = None, ttl: int = None, wait: bool = True, timeout: float = None) → LockId[source]
abstract async release(id: NSKey, identifier: LockId) → None[source]
abstract async owner(id: NSKey) → LockId | None[source]
abstract async is_owner(id: NSKey) → bool[source]
abstract async m_exists(id: Collection[NSKey]) → frozenset[kaiju_tools.types.NSKey][source]
class RPCServer[source]

Bases: ABC

JSONRPC server interface.

abstract async call(body, headers: Mapping, callback: Callable[[...], Awaitable] = None) → tuple[source]
class RPCClient[source]

Bases: ABC

JSONRPC client interface.

abstract async call(method: str, params: dict | None = None, nowait: bool = False, request_id: int = 0, max_timeout: int = None, use_context: bool = True, retries: int = None, headers: dict = None) → Optional[source]
abstract async call_multiple(requests, raise_exception: bool = True, nowait: bool = False, max_timeout: int = None, use_context: bool = True, retries: int = None, abort_on_error: bool = None, use_template: bool = None, headers: dict = None) → Optional[source]
class ServiceManagerInterface[source]

Bases: ABC

Application service initializer interface.

abstract add_service(service, required: bool = True, name: str = None) → None[source]
abstract discover_service(name=None, cls=None, required: bool = True) → Optional[source]
abstract items() → Iterable[source]
class App[source]

Bases: Application, Generic[_Session]

Web application.

id: str

Unique instance id (auto-generated).

name: str

Application name (i.e. type).

version: str

App version.

env: str

Current environment: dev, prod, etc.

loglevel: str

Root log level.

logger: Logger

Application root logger.

services: ServiceManagerInterface

Services map.

settings: dict

Application settings.

namespace: Namespace

Application namespace.

namespace_shared: Namespace

Shared namespace for the current environment.

request_context: ContextVar[RequestContext | None]

Client request context is stored here.

request_session: ContextVar[_Session | None]

Client session context is stored here.

db_meta: Any

Database metadata.

cookie_key: str

HTTP cookie key for client sessions.

class TokenLoginInterface[source]

Bases: ABC

Token client interface.

abstract async get_token() → TokenInfo[source]
class Stream[source]

Bases: ABC

Streaming interface.

abstract async lock() → None[source]
abstract async unlock() → None[source]
abstract property locked: bool