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

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

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

create_named_entity_view()[source]

Create a vue for entities with names

persons, objects and geoentities

create_nfunction_view()[source]

Create a vue that links people to acts through functions

TODO: this should be generalized to objects (named entities)

create_pattribute_view()[source]

Return the pattribute 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)

create_tables()[source]

Creates the tables from the current ORM metadata if needed

Returns:: None

create_views()[source]

Creates the views

eattributes: view of the entity attributes linked with entity information pattributes: view of the person attributes linked with person information named_entities: view of the named entities linked with entity information nfunctions: view of the functions of people in acts linked with entity information See issue #63 :return: None

describe(argument, show=None, **kwargs)[source]

Describe a table or a model: if argument is a string, it is assumed to be a table 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:

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]) – 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], 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

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.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()