timelink.api package

Subpackages

Submodules

timelink.api.crud module

CRUD operations for the timelink API.

timelink.api.crud.get(db: Session, id: str) → EntityAttrRelSchema[source]

Get entity by id :param db: database session :param id: entity id

Returns:

Entity object

timelink.api.crud.get_syslog(db: Session, nlogs: int) → list[SysLog][source]

Get last n system logs last one first :param db: database session :param nlogs: sequence number

Returns:

List of SysLog objects

timelink.api.crud.get_syslog_by_time(db: Session, start_time: datetime, end_time: datetime) → list[SysLog][source]

Get system logs between start_time and end_time :param db: database session :param start_time: start time :param end_time: end time

Returns:

List of SysLog objects

timelink.api.crud.get_syspar(db: Session, q: list[str] | None = None)[source]

Get system parameters :param db: database session :param q: parameter name(s); if empty, return all parameters

Returns:

SysPar object

timelink.api.crud.set_syslog(db: Session, log: SysLogCreateSchema) → SysLog[source]

Set system log :param db: database session :param log: SysLogCreateSchema object with level, origin and message

Returns:

SysLog object

timelink.api.crud.set_syspar(db: Session, syspar: SysParSchema)[source]

Set system parameters :param db: database session :param syspar: SysPar object

Returns:

SysPar object

timelink.api.database module

Database connection and setup

TODO:

  • use logging to database functions.

  • add mysql support

class timelink.api.database.TimelinkDatabase(db_name: str = 'timelink', db_type: str = 'sqlite', db_url=None, db_user=None, db_pwd=None, db_path=None, kleio_server=None, kleio_home=None, kleio_image=None, kleio_version=None, kleio_token=None, kleio_update=None, postgres_image=None, postgres_version=None, stop_duplicates=True, echo=False, **connect_args)[source]

Database connection and setup

Creates a database connection and session. If the database does not exist, it is created. db_type determines the type of database.

Currently, only sqlite and postgres are supported.

  • If db_type is sqlite, the database is created in the current directory.

  • If db_type is postgres or mysql, the database is created in a docker container.

  • If the database is postgres, the container is named timelink-postgres.

  • If the database is mysql, the container is named timelink-mysql.

db_url

database sqlalchemy url

Type:

str

db_name

database name

Type:

str

db_user

database user (only for postgres databases)

Type:

str

db_pwd

database password (only for postgres databases)

Type:

str

engine

database engine

Type:

Engine

session

database session factory

Type:

Session

metadata

database metadata

Type:

MetaData

db_container

database container

kserver

kleio server attached to this database, used for imports

Type:

timelink.kleio.kleio_server.KleioServer | None

Main methods:

  • table_names: get the current tables in the database

  • get_columns: get the columns for a table or model

  • table_row_count: get the number of rows of each table in the database

  • get_models: get ORM Models for using in Queries

  • create_db: create the database tables and views

  • drop_db: drop all timelink related tables from the database

  • get_imported_files: get the list of imported files in the database

  • get_import_status: get the import status of the kleio files

  • update_from_sources: update the database from the sources

  • query: execute a query in the database

  • get_person: fetch a person by id

  • get_entity: fetch an entity by id

  • export_as_kleio: export entities to a kleio file

Initialize the database connection and setup

Example

db = TimelinkDatabase(‘timelink’, ‘sqlite’) with db.session() as session:

# do something with the session session.commit()

Parameters:

  • db_name (str, optional) – database name; defaults to “timelink”.

  • db_type (str, optional) – database type; defaults to “sqlite”.

  • db_url (str, optional) – database url. If None, a url is generated; defaults to None

  • db_user (str, optional) – database user; defaults to None.

  • db_pwd (str, optional) – database password; defaults to None.

  • db_path (str, optional) – database path (for sqlite databases); defaults to None.

  • kleio_server (KleioServer, optional) – kleio server for imports; defaults to None.

  • kleio_home (str, optional) – kleio home directory; defaults to None. If present and kleio_server is None will start new kleio server, which can be fetched with get_kleio_server()

  • kleio_image (str, optional) – kleio docker image. Passed to KleioServer().

  • kleio_version (str, optional) – kleio version. Passed to KleioServer().

  • kleio_token (str, optional) – kleio token. Passed to KleioServer().

  • kleio_update (bool, optional) – update kleio server. Passed to KleioServer().

  • postgres_image (str, optional) – postgres docker image; defaults to None.

  • postgres_version (str, optional) – postgres version; defaults to None.

  • echo (bool, optional) – if True, the Engine will log all statements; defaults to False.

  • connect_args (dict, optional) – extra arguments to sqlalchemy and timelink.kleio.KleioServer.start()

create_db()[source]

Create the database

Will create the tables and views if they do not exist Will load the database classes and ensure all mappings Will stamp the database with alembic as most recent version

db_table_names()[source]

Current tables in the database

describe(argument, show=None, **kwargs)[source]
Describe a table, view or a model

if argument is a string, it is assumed to be a table or view if argument is a model, it is assumed to be a ORM model otherwise it is checked if it is a table object the method prints the columns of the table or model

Parameters:

  • argument – table name or model

  • kwargs – additional arguments to pass to the describe method

  • show – print the columns

Returns:

list of columns

Return type:

list

drop_db(session=None)[source]

This will drop all timelink related tables from the database. It will not touch non timelink tables that might exist.

If a real drop database is desidered use sqlalchemy_utils.drop_database :param session:

export_as_kleio(ids: List, filename, kleio_group: str | None = None, source_group: str | None = None, act_group: str | None = None)[source]

Export entities to a kleio file

Renders each of the entities in the list in kleio format using Entity.to_kleio() and writes them to a file.

If provded, kleio_group, source_group and act_group are written before the entities.

Parameters:

  • ids (List) – list of ids

  • filename ([type]) – destination file path

  • kleio_group ([type]) – initial kleio group

  • source_group ([type]) – source group

  • act_group ([type]) – act group

get_columns(class_or_table: str)[source]

Get the columns for a entity type

Returns:

list of columns

Return type:

list

get_database_version()[source]

Get the alembic version string for this db

get_db()[source]

Get a database session :returns: database session :rtype: Session

get_engine()[source]

Get the database engine :returns: database engine :rtype: Engine

get_entity(id: str, session=None) → Entity[source]

Fetch an entity by id.

See: timelink.api.models.entity.Entity.get_entity()

get_import_rpt(path: str, match_path=False) → str[source]

Get the import report of a file in the database

get_import_status(kleio_files: List[KleioFile] | None = None, path=None, recurse=True, status=None, match_path=False) → List[KleioFile][source]

Get the import status of the kleio files

The import status is stored in the database. This method retrieves the import status of the kleio files.

Parameters:

  • kleio_files (List[KleioFile], optional) – list of kleio files; defaults to None.

  • path (str, optional) – path to the sources; defaults to None.

  • recurse (bool, optional) – if True, recurse the path; defaults to True.

  • status (import_status_enum, optional) – import status; defaults to None.

  • math_path (bool, optional) – if True, match the path of the kleio file with the path of the imported file; defaults to False (match just the file name).

See timelink.api.database.get_import_status()

get_imported_files() → List[KleioImportedFileSchema][source]

Returns the list of imported files in the database.

get_kleio_server()[source]

Return the kleio server associated with this database

get_metadata()[source]

Get the database metadata :returns: database metadata :rtype: MetaData

get_model(class_id: str | List[str], make_alias=None)[source]

Get the ORM class for a entity type

Parameters:

  • class_id (str | List[str]) – class id or list of class ids

  • make_alias (bool, optional) – if True, return an aliased ORM class; defaults to True in lists; False if single class_id.

Returns:

ORM class

get_model_by_name(class_or_groupname: str, make_alias=False)[source]

Get the ORM class for a entity type by name or for a group name. If the name is not found, return None

Parameters:

class_or_groupname (str) – class or group name

Returns:

//docs.sqlalchemy.org/en/20/errors.html#error-xaj2

Return type:

ORM class aliased to avoid # https

get_models_ids()[source]

Get the ORM model classes as a list of ids

Returns:

list of ORM classes as string ids

Return type:

list

get_need_import(kleio_files: List[KleioFile] | None = None, with_import_errors=False, with_import_warnings=False, match_path=False) → List[KleioFile][source]

Get the kleio files that need import

These are the files that were never imported or that were updated after import (status N and U). Use include_errors and include_warnings set to true to also include files previously imported with errors and warnings.

Parameters:

  • kleio_files (List[KleioFile]) – list of kleio files

  • with_import_errors (bool, optional) – if True, include files with errors; defaults to False.

  • with_import_warnings (bool, optional) – if True, include files with warnings; defaults to False.

  • match_path (bool, optional) – if True, match the path of the kleio file with the path of the imported file; defaults to False.

Returns:

list of kleio files with field import_status

Return type:

List[KleioFile]

get_person(*args, **kwargs)[source]

Fetch a person by id

See timelink.api.models.person.get_person()

get_table(table_or_class: str | Entity) → Table[source]

Get a table object from the database

Parameters:

table_or_class (str | Entity) – table name, model name of ORM model

Returns:

table object

Return type:

sqlAlchemy Table

get_view(view_name: str)[source]

Get a view by name

get_view_columns(view_name: str)[source]

Get the columns of a view

import_from_xml(file: str | KleioFile, kserver=None, return_stats=True)[source]

Import one file

Parameters:

file (str | KleioFile) – path to xml file or KleioFile object kserver (KleioServer, optional): Kleio server to use for import. Defaults to None. return_stats (bool, optional): Return import stats. Defaults to True.

orm_table_names()[source]

Current tables associated with ORM models

pperson(id: str, session=None)[source]

Prints a person in kleio notation

query(query_spec)[source]

Executes a query in the database

Parameters:

query – sqlAlchemy query

set_kleio_server(kleio_server: KleioServer)[source]

Set the kleio server for imports

Parameters:

kleio_server (KleioServer) – kleio server

table_row_count() → List[tuple[str, int]][source]

Number of rows of each table in the database

Returns:

list of tuples (table_name, row_count)

Return type:

list

update_from_sources(path=None, recurse=True, with_translation_warnings=True, with_translation_errors=False, with_import_errors=False, with_import_warnings=False, force=True, match_path=False)[source]

Update the database from the sources.

Needs an attached KleioServer.

Parameters:

  • path (str) – path to the sources, if None all the sources are updated.

  • recurse (bool, optional) – if True, recurse the path; defaults to True.

  • with_translation_warnings (bool, optional) – if True, import files with translation warnings; defaults to True.

  • with_translation_errors (bool, optional) – if True, import files with tr errors; defaults to False.

  • with_import_errors (bool, optional) – if True, re-import files with errors; defaults to False.

  • with_import_warnings (bool, optional) – if True, re-import files with warnings; defaults to False.

  • match_path (bool, optional) – if True, match the path of the kleio file with the path of the imported file; if False just match the file name; defaults to False.

  • force (bool) – force the translation and subsequent import of valid files

view_names()[source]

Get the list of views in the database

timelink.api.schemas module

Schemas for the Timelink API

In the FastAPi tutorial this file is used for the pydantic models for the API, including for the classes that are used for database access. Another file, called models.py, is used in the tutorial for the SQLAlchemy models.

We use a module called “models” for the SQLAlchemy models

Here we put the pydantic models that are not related to database models like search requests and search results.

class timelink.api.schemas.AttributeSchema(*, entity: str, the_type: str, the_value: str, the_date: str, obs: str | None, groupname: str = None)[source]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.EntityAttrRelSchema(**data: Any)[source]

Pydantic Schema for Entity with attributes, relation and contained entities

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.EntityBriefSchema(*, id: str, pom_class: str, inside: str | None, the_source: str | None, the_order: int | None, the_level: int | None, the_line: int | None, groupname: str | None, extra_info: dict | None, updated: datetime | None, indexed: datetime | None)[source]

Pydantic Schema for Entity brief

No links to other entities

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.EntitySchema(*, id: str, pom_class: str, inside: str | None, the_source: str | None, the_order: int | None, the_level: int | None, the_line: int | None, groupname: str | None, updated: datetime | None, indexed: datetime | None, extra_info: dict | None, contains: List[EntitySchema] | None)[source]

Pydantic Schema for Entity

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.ImportStats(*, datetime: date, machine: str, database: str, file: str, import_time_seconds: float, entities_processed: int, entity_rate: float, person_rate: float, nerrors: int, errors: List[str])[source]

Import statistics

Fields:

datetime: date and time of import machine: machine where import was done database: specific database where import was done file: file that was imported import_time_seconds: time in seconds that import took entities_processed: number of entities processed entity_rate: number of entities processed per second person_rate: number of persons processed per second nerrors: number of errors errors: list of errors

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.LinkSchema(*, id: int, rid: str, entity: str, user: str, rule: str, source: str, status: LinkStatus, aregister: str | None)[source]

Identification links between entities

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.RealEntityAttrRelSchema(**data: Any)[source]

Pydantic Schema for RealEntity

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.RealEntitySchema(**data: Any)[source]

Pydantic Schema for RealEntity

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.RelationInSchema(*, id: str, origin: str, destination: str, the_type: str, the_value: str, the_date: str, obs: str | None, org_name: str | None)[source]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.RelationOutSchema(*, id: str, origin: str, destination: str, the_type: str, the_value: str, the_date: str, obs: str | None, dest_name: str | None)[source]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.RelationSchema(*, id: str, origin: str, destination: str, the_type: str, the_value: str, the_date: str, obs: str | None)[source]

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {'from_attributes': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.SearchRequest(*, q: str, after: date | None = None, until: date | None = None, skip: int | None = 0, limit: int | None = 100)[source]

Search request

Fields:

q: search query after: date after which to search, possibly None until: date until which to search, possibly None skip: number of items to skip, default 0 limit: number of items to return, default 100

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class timelink.api.schemas.SearchResults(*, id: str, the_class: str, description: str, start_date: date, end_date: date)[source]

Search results

Fields:

results: list of search results

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

timelink.api.views module

class timelink.api.views.CreateView(name, selectable)[source]
class timelink.api.views.DropView(name)[source]