gonk.impl.sq3

class gonk.impl.sq3.RecordKeeper(parent_directory: Path)

SQLite backed RecordKeeper.

Events are stored in a table, JSON serialized. ID primary key and event UUID are indexed.

add(event: EventT)

Add a gonk.core.events.Event to storage.

read(uuid_: UUID) → Event

Read a gonk.core.events.Event from storage.

Raises:

ValueError – Event does not exist.

exists(uuid_: UUID) → bool

Check gonk.core.events.Event existence.

next(uuid_: UUID | None = None) → UUID | None

Get the next gonk.core.events.Event’s

UUID in the event history.

Returns:

The next event UUID in the event history

or None if the provided UUID is the tail.

Raises:

ValueError – Event does not exist.

tail() → UUID | None

Get the last gonk.core.events.Event’s

UUID in the event history.

Returns:

The last event UUID in the event history

or None when there are no events.

class gonk.impl.sq3.StatusT(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Enum for tracking object and annotation status in state.

No status implies create accapted.

CREATE_PENDING = 1

Create event pending.

CREATE_REJECTED = 2

Create event rejected.

DELETE_PENDING = 4

Delete event pending.

DELETE_ACCEPTED = 8

Delete event accepted.

class gonk.impl.sq3.State(parent_directory: Path, record_keeper: RecordKeeper)

SQLite backed State.

See parent gonk.core.interfaces.State for more information.

events_by_object(identifier: Identifier)

Get the events for a given object.

events_by_annotation(identifier: Identifier)

Get the events for a given annotation.

events_all(after: None | UUID = None)

Get a page of 25 events.

Parameters:

after – Event UUID to page after or None for the first page.

annotations_all(uuid_: None | UUID = None, after: None | UUID = None)

Get a page of 25 annotations.

Only one of uuid_ and after may be provided.

Parameters:

• uuid – The annotation UUID to get information about.

• after – Annotation UUID to page after or None for the first page.

annotations_by_object(object_identifier: Identifier)

Gets the annotations for a given object.

annotations_by_status(status: str, after: None | UUID = None)

Gets the annotations for a given status.

Parameters:

• status – pending, accepted, rejected, or deleted.

• after – Annotation UUID to page after or None for the first page.

annotation(identifier: Identifier)

Get an Annotation for a given identifier.

objects_all(uuid_: None | UUID = None, after: None | UUID = None)

Get a page of 25 objects.

Only one of uuid_ and after may be provided.

Parameters:

• uuid – The object UUID to get information about.

• after – Object UUID to page after or None for the first page.

objects_by_annotation(annotation_uuid: UUID)

Get the objects for a given annotation.

objects_by_status(status: str, after: None | UUID = None)

Gets the objects for a given status.

Parameters:

• status – pending, accepted, rejected, or deleted.

• after – Object UUID to page after or None for the first page.

object(identifier: Identifier)

Get an Object for a given identifier.

schemas_all(name: None | str = None)

Get all schemas.

Parameters:

name – If provided only information about that schema will be returned.

schemas_by_status(status: str, after: None | UUID = None)

Gets the schemas for a given status.

Parameters:

• status – pending, accepted, rejected, or deprecated.

• after – Schema UUID to page after or None for the first page.

schema(name: str, version: int) → None | Object

Get a schema for a given name and version.

owners()

Get all owners.

consume(event: EventT)

Dispatch method for event consumption methods.