titanfe.repository module

Repository can be used to connect to the titan repository service

class titanfe.repository.RepositoryService(logger, reposervice_address=None)

Bases: object

Repository service implements a connection to the titan repository service

Parameters:

• brick_name (string) – the name of the brick instance

• logger – the logger instance of the parent

• repo_service (string) – optional, address of the repository service

delete(collection, document)

delete data using the repository service

get(collection, document, find)

get data using the repository service

store(collection, document, value)

store data using the repository service

class titanfe.repository.Request(address: str, method: Callable, content: titanfe.repository.RequestData, log: logging.LoggerAdapter, response: Any = <factory>)

Bases: object

Request object

Parameters:

• address – str Target address

• method – Callable requests method(get, put, delete..)

• logger – the logger instance of the parent

• content – RequestData request content to be sent

address: str

content: titanfe.repository.RequestData

log: logging.LoggerAdapter

method: Callable

response: Any

send()

send request

class titanfe.repository.RequestData(collection: str, document: str, value: Any, find: Optional[dict] = <factory>, database: str = 'BrickRunner')

Bases: object

Request data object sent to the Repository service

collection: str

database: str = 'BrickRunner'

document: str

find: Optional[dict]

classmethod from_dict(kvs: Optional[Union[dict, list, str, int, float, bool]], *, infer_missing=False) → A

classmethod from_json(s: Union[str, bytes, bytearray], *, parse_float=None, parse_int=None, parse_constant=None, infer_missing=False, **kw) → A

classmethod schema(*, infer_missing: bool = False, only=None, exclude=(), many: bool = False, context=None, load_only=(), dump_only=(), partial: bool = False, unknown=None) → dataclasses_json.mm.SchemaF[A]

to_dict(encode_json=False) → Dict[str, Optional[Union[dict, list, str, int, float, bool]]]

to_json(*, skipkeys: bool = False, ensure_ascii: bool = True, check_circular: bool = True, allow_nan: bool = True, indent: Optional[Union[int, str]] = None, separators: Tuple[str, str] = None, default: Callable = None, sort_keys: bool = False, **kw) → str

value: Any