timelink.api package

Subpackages

timelink.api.models package

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, **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

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.
extra_args (dict, optional) – extra arguments to sqlalchemy and timelink.kleio.KleioServer.start()

create_tables()[source]

Creates the tables from the current ORM metadata if needed

Returns:: None

create_views()[source]

Creates the views

Returns:: None

drop_db(session)[source]: This will drop all timelink related tables from the database. It will not touch non timelink tables that might exist. :param session: :return:

ensure_all_mappings(session)[source]: Ensure that all database classes have a table and ORM class

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]) – text to be used as initial kleio group
source_group ([type]) – text to be used as source group
act_group ([type]) – text to be used as act group

get_db()[source]: Get a database session :returns: database session :rtype: Session

get_eattribute_view()[source]

Return the eattribute view.

Returns a sqlalchemy table with a view that joins the table “entities” and the table “attributes”. This view provides attribute values with the “positional” information kept in the entities tables, such as line number, level and order in the source file as well as groupname of the attribute (“ls”, “attr”, etc…) and timestamps for updates and indexing.

get_engine()[source]: Get the database engine :returns: database engine :rtype: Engine

get_entity(*args, **kwargs) → 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])[source]

Get the ORM class for a entity type

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], 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_pattribute_view()[source]

Return the nattribute view.

Returns a sqlalchemy table linked to the pattributes view of timelink/MHK databases This views joins the table “persons” and the table “attributes” providing attribute values with person names and sex.

The column id contains the id of the person/object, not of the attribute

Returns:: A table object with the pattribute view

Original SQL code

CREATE VIEW pattributes AS
    SELECT p.id        AS id,
        p.name      AS name,
        p.sex       AS sex,
        a.the_type  AS the_type,
        a.the_value AS the_value,
        a.the_date  AS the_date,
        p.obs       AS pobs,
        a.obs       AS aobs
    FROM attributes a, persons p
    WHERE (a.entity = p.id)

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

Fetch a person by id

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

get_table(class_id: str)[source]

Get the ORM table for a entity type

Returns:: Table

load_database_classes(session)[source]: Populates database with core Database classes :param session: :return:

pperson(id: str)[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_names()[source]: Current tables in the database

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, 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.

timelink.api.database.get_import_status(db: TimelinkDatabase, kleio_files: List[KleioFile], match_path=False) → List[KleioImportedFileSchema][source]

Get the import status of the kleio files.

The status in returned in timelink.api.models.system.KleioImportedFileSchema.import_status

Parameters:

db (TimelinkDatabase) – timelink database
kleio_files (List[KleioFile]) – list of kleio files with extra field import_status
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[KleioImportedFileSchema]

TODO: KleioFile should also receive the import_errors and import_warnings

timelink.api.database.get_postgres_container() → Container[source]: Get the postgres container :returns: the postgres container :rtype: docker.models.containers.Container

timelink.api.database.get_postgres_container_pwd() → str[source]: Get the postgres container password :returns: the postgres container password :rtype: str

timelink.api.database.get_postgres_container_user() → str[source]: Get the postgres container user :returns: the postgres container user :rtype: str

timelink.api.database.get_postgres_dbnames()[source]

Get the database names from a postgres server :returns: list of database names :rtype: list[str]

..code-block:: sql

SELECT datname
FROM pg_database

WHERE NOT datistemplate
AND datallowconn AND datname <> ‘postgres’;

timelink.api.database.get_sqlite_databases(directory_path: str) → list[str][source]

Get the sqlite databases in a directory :param directory_path: directory path :type directory_path: str

Returns:: list of sqlite databases
Return type:: list[str]

timelink.api.database.is_postgres_running()[source]: Check if postgres is running in docker

timelink.api.database.random_password()[source]: Generate a random password

timelink.api.database.start_postgres_server(dbname: str | None = 'timelink', dbuser: str | None = 'timelink', dbpass: str | None = None, image: str | None = 'postgres', version: str | None = 'latest')[source]: Starts a postgres server in docker :param dbname: database name :type dbname: str :param dbuser: database user :type dbuser: str :param dbpass: database password :type dbpass: str :param version: postgres version; defaults to “latest”. :type version: str | None, optional

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].

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].

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].

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.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

Handling views from sql Alchemy

View utilities from https://github.com/sqlalchemy/sqlalchemy/wiki/Views

class timelink.api.views.CreateView(name, selectable)[source]: Create a View

class timelink.api.views.DropView(name)[source]: Drop a View

timelink.api.views.view(name, metadata, selectable)[source]

Create a view with the given name from the given selectable.

The view is created when the metadata is first bound to an engine.

Example:

stuff_view
     = view("stuff_view", metadata, sa.select(
         stuff.c.id.label("id"),
         stuff.c.data.label("data"),
         more_stuff.c.data.label("moredata"),
     )
         .select_from(stuff.join(more_stuff))
         .where(stuff.c.data.like(("%orange%"))),
     )

 with engine.connect() as conn:
 conn.execute(
     sa.select(stuff_view.c.data, stuff_view.c.moredata)
 ).all()