types - basic types

This module contains a collection of various basic types and classes used throughout the package.

class SortedStack

Bases: Sized, Iterable

A sorted collection (stack) of items.

>>> stack = SortedStack({'dogs': 12, 'sobaki': 5})
>>> stack = SortedStack(stack)
>>> stack.extend(SortedStack({'cats': 5}))

Selection:

>>> stack.select(8)
['sobaki', 'cats']

>>> stack.rselect(8)
['dogs']

Insertion and removal:

>>> stack.insert(1, 'koty')
>>> stack.pop_many(3)
['koty']

>>> stack.pop()
'sobaki'

>>> len(stack)
2

>>> stack.clear()
>>> bool(stack)
False

__init__(items: Iterable[tuple] | dict = None, /)

Initialize.

property lowest_score

Get the lowest score in the stack.

extend(items: Iterable, /) → None

Extend the stack by adding more than one element.

insert(score, item, /) → None

Insert a single element into the stack.

select(score_threshold, /) → list

Select and return items without removing them from the lowest score to score_threshold.

The values are guaranteed to be in order.

rselect(score_threshold, /) → list

Select and return items without removing them from the highest score to score_threshold.

The values are guaranteed to be in order.

pop()

Pop a single element which has the lowest score.

Raises:

StopIteration – if there are no values to return.

rpop()

Pop a single element which has the highest score.

Raises:

StopIteration – if there are no values to return.

pop_many(score_threshold, /) → list

Pop and return values with scores less than score_threshold.

The returned values are guaranteed to be in order. Returns an empty list if no values.

rpop_many(score_threshold, /) → list

Pop and return values with scores greater than score_threshold.

Returned values are guaranteed to be in order.

clear() → None

Clear all values.

class RestrictedDict

Bases: dict

Same as a normal mapping but forbids key updates.

>>> m = RestrictedDict(a=1, b=2)
>>> m['c'] = 3
>>> 'c' in m
True

Resetting a key raises a ValueError:

>>> m['c'] = 5  
Traceback (most recent call last):
...
ValueError:

>>> m.update({'c': 5})   
Traceback (most recent call last):
...
NotImplementedError:

So as trying to remove it:

>>> m.pop('c')  
Traceback (most recent call last):
...
NotImplementedError:

class Namespace

Bases: Mapping

Namespace can be used for shared key and name management.

Create namespaces.

>>> ns = Namespace(env='dev', name='app')
>>> sub_ns = ns / 'sub'  # eq to .create_namespace('sub')
>>> str(sub_ns)
'dev.app.sub'

Get keys.

>>> sub_ns.get_key('key')
'dev.app.sub.key'

env: str

Environment: dev, prod, etc.

name: str

Application or scope name.

namespaces: dict[str, kaiju_tools.types.Namespace]

Sub-namespaces.

get_key(_Namespace__key: str) → NSKey

Get a shared key.

create_namespace(_Namespace__ns: str) → Namespace

Create a new sub-namespace.

classmethod join(*args)

Join arguments in a namespace key.

You should use NSKey.get_key instead of this when possible.

__init__(env: str, name: str) → None

class RequestContext

Bases: TypedDict

Request context for an RPC request.

Each request going through the RPC server has a unique context assigned to the context variable.

correlation_id: str

request chain correlation id header

session_id: str | None

user session id

request_deadline: int | None

unix time deadline for this request chain

data: dict | None

arbitrary user data

class Session

Bases: Serializable

User session data.

__init__(*, id: str, h_agent: bytes | None, user_id: UUID | None, expires: int, permissions: frozenset[str], data: dict, created: datetime, _stored: bool, _changed: bool, _loaded: bool)

Initialize.

Parameters:

• id – session id

• h_agent – user agent hash

• user_id – user id in users table

• expires – unix timestamp when this session should expire

• permissions – a set of session-bound permissions

• data – additional stored data

• created – timestamp

user_id

current user (or None)

expires

expiration unix timestamp

permissions

a set of user permissions

data

session data

property scope: Scope

Base user scope.

property stored: bool

Session should be stored.

property changed: bool

Session has changed.

property loaded: bool

Session has been loaded from db.

update(data: dict)

Update session data.

clear()

Clear all session data.

class Scope

Session permission scopes for RPC methods.

SYSTEM = 0

System scope: only internal and system services can access.

ADMIN = 10

Administrators can access.

USER = 100

Authenticated users can access.

GUEST = 1000

No access restrictions.