timelink.api package

Subpackages

Submodules

timelink.api.crud module

CRUD operations for the Timelink API.

This module provides basic Create, Read, Update, and Delete operations for Timelink system models, including system parameters, logs, and general entities.

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]

Retrieve the last n system logs, most recent first.

Parameters:

  • db (Session) – Database session.

  • nlogs (int) – Number of log entries to retrieve.

Returns:

A list of system log objects.

Return type:

list[SysLog]

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

Retrieve system logs within a specific time range.

Parameters:

  • db (Session) – Database session.

  • start_time (datetime) – Start of the time range.

  • end_time (datetime) – End of the time range.

Returns:

A list of system log objects within the specified range.

Return type:

list[SysLog]

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

Retrieve system parameters from the database.

Parameters:

  • db (Session) – Database session.

  • q (list[str] | None, optional) – List of parameter names to retrieve. If None, returns all parameters. Defaults to None.

Returns:

A list of system parameter objects.

Return type:

list[SysPar]

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

Create a new system log entry.

Parameters:

  • db (Session) – Database session.

  • log (SysLogCreateSchema) – Pydantic schema containing level, origin, and message.

Returns:

The created system log object.

Return type:

SysLog

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

Create or update a system parameter.

Parameters:

  • db (Session) – Database session.

  • syspar (SysParSchema) – Pydantic schema containing parameter data.

Returns:

The created or updated system parameter object.

Return type:

SysPar

timelink.api.database module

Database connection and setup

This module provides the TimelinkDatabase class, which manages the connection to a Timelink database and provides high-level methods for database operations. The class integrates several mixins to provide specialized functionality for PostgreSQL, SQLite, Kleio server integration, metadata inspection, and querying.

class timelink.api.database.KleioFile(*, path: str, name: str, size: int, directory: str, modified: datetime, modified_iso: datetime, modified_string: str, qtime: datetime, qtime_string: str, source_url: str, status: translation_status_enum, translated: datetime | None = None, translated_string: str | None = None, errors: int | None = None, warnings: int | None = None, version: str | None = None, rpt_url: str | None = None, xml_url: str | None = None, import_status: import_status_enum | None = None, import_errors: int | None = None, import_warnings: int | None = None, import_error_rpt: str | None = None, import_warning_rpt: str | None = None, imported: datetime | None = None, imported_string: str | None = None)[source]

Represents the information about a kleio file and its translation and import status.

path

The path of the file.

Type:

str

name

The name of the file.

Type:

str

size

The size of the file in bytes.

Type:

int

directory

The directory containing the file.

Type:

str

modified

The last modified time of the file.

Type:

datetime

modified_iso

The last modified time of the file in ISO format.

Type:

datetime

modified_string

The last modified time of the file as a string.

Type:

str

qtime

The time the file was queued for translation.

Type:

datetime

qtime_string

The time the file was queued for translation as a string.

Type:

str

source_url

The URL of the source file.

Type:

str

status

The status of the file: - V: valid translations - T: need translation (source more recent than translation) - E: translation with errors - W: translation with warnings - P: translation being processed - Q: file queued for translation

Type:

translation_status_enum

translated

The time the file was translated.

Type:

Optional[datetime]

translated_string

The time the file was translated as a string.

Type:

Optional[str]

errors

The number of errors encountered during translation.

Type:

Optional[int]

warnings

The number of warnings encountered during translation.

Type:

Optional[int]

version

The version of the kleio translator.

Type:

Optional[str]

rpt_url

The URL of the report file.

Type:

Optional[str]

xml_url

The URL of the XML file.

Type:

Optional[str]

import_status

The status of the file import: - I: imported - E: imported with error - W: imported with warnings no errors - N: not imported - U: translation updated need to reimport

Type:

Optional[import_status_enum]

import_errors

The number of errors encountered during import.

Type:

Optional[int]

import_warnings

The number of warnings encountered during import.

Type:

Optional[int]

import_error_rpt

Error report from import.

Type:

Optional[str]

import_warning_rpt

Warning report from import.

Type:

Optional[str]

imported

Date of import of the file.

Type:

Optional[datetime]

imported_string

Date of import of the file as a string.

Type:

Optional[str]

needs_translation()[source]

Return True if the file needs translation.

needs_import()[source]

Return True if the file needs import. A file needs import if it has not been imported before or if it has been translated again since the last import.

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

needs_import()[source]

Return True if the file needs import

A file needs import if it has not been imported before

or if it has been translated again since the last import

needs_translation()[source]

Return True if the file needs translation

source_url: str

The status of the file: .. code-block:: python

V = “valid translations” T = “need translation (source more recent than translation)” E = “translation with errors” W = “translation with warnings” P = “translation being processed” Q = “file queued for translation”

class timelink.api.database.KleioServer(container: Container | None = None, url: str | None = None, token: str | None = None, kleio_home: str | None = None)[source]

This class interfaces to a Kleio server through its JSON-RPC api. It also provides convenience methods to start a server in Docker locally.

This class is not intended to be instantiated directly. Use KleioServer.start() and KleioServer.attach() to create instances of KleioServer.

Parameters:

  • container (docker.models.containers.Container) – runing kleio server container

  • url (str) – kleio server url if running in a different machine (container=None)

  • token (str) – kleio server token if running in a different machine (container=None)

  • kleio_home (str) – kleio server home directory. If None and container is not None, then kleio_home is obtained from the container.

Main methods:

  • start: Starts a kleio server in docker.

  • attach: Attach to an already running Kleio Server.

  • get_server: Return a running kleio server associated with a kleio_hom

  • is_server_running: Check if a kleio server is running in docker mapped to a given kleio home directory.

  • find_local_kleio_home: Find kleio home directory.

  • make_token: Get the kleio server token from environment or generate a new one if not set.

  • get_token: Get the running kleio server token.

  • get_kleio_home: Get the kleio server home directory.

  • get_container: Get the kleio server container.

  • get_logs: Get the logs of the kleio server container.

  • get_url: Get the kleio server url.

  • call: Basic call to kleio server API.

  • stop: Stop the kleio server container.

  • invalidate_user: Invalidate a user.

  • generate_token: Generate a token for a user.

  • get_translations: Get translation status from kleio server.

  • translate: Translate sources from kleio server.

  • translation_clean: Clean translations from kleio server.

  • get_sources: Get sources from kleio server.

  • get_report: Get report from kleio server.

  • get_url_content: Get content from Kleio Server.

  • get_home_page: Get home page from Kleio Server.

Not to be used directly.

See KleioServer.start() and KleioServer.attach()

To start a kleio server locally in docker use KleioServer.start()

To attach to a running kleio server, local or remote, use KleioServer.attach()

Parameters:

  • container (docker.models.containers.Container) – runing kleio server container; if container is None, then url, token and kleio_home must be provided.

  • url (str) – kleio server url if running in a different machine (container=None)

  • token (str) – kleio server token if running in a different machine (container=None)

  • kleio_home (str) – kleio server home directory. If None and container is not None then kleio_home is obtained from the container. If not none

static attach(url: str, token: str, kleio_home: str = '.')[source]

Attach to a already running Kleio Server.

Use this either to attach to a running server outside docker (for instance in a Prolog session) or to a server running in another machine (should use htpps url in that case)

Parameters:

  • url (str) – kleio server url

  • token (str) – kleio server token

  • kleio_home (str, optional) – kleio server home directory

Returns:

KleioServer object

Return type:

KleioServer

TODO #33: from urllib.parse import urlparse

url = ‘http://user:token@localhost:8088/path_to_kleio_home’ parsed_url = urlparse(url)

print(‘scheme:’, parsed_url.scheme) print(‘netloc:’, parsed_url.netloc) print(‘path:’, parsed_url.path) print(‘username:’, parsed_url.username) print(‘password:’, parsed_url.password) print(‘hostname:’, parsed_url.hostname) print(‘port:’, parsed_url.port)

call(method: str, params: dict, token: str | None = None, timeout=60)[source]

Call kleio server API

Parameters:

  • method (str) – kleio server API method

  • params (dict) – kleio server API method parameters

  • token (str, optional) – kleio server token; defaults to None -> use admin token

  • timeout (int, optional) – timeout in seconds; defaults to 60.

Returns:

kleio server API response

Return type:

dict

container: Container

kleio server container

extract_version_info(home_page_content: str) → Tuple[str, str, str] | None[source]

Extract version information from the home page content.

Parameters:

home_page_content (str) – The content of the home page.

Returns:

A tuple containing the version number, build number, and date of build,

or None if the information could not be extracted.

Return type:

Optional[Tuple[str, str, str]]

static find_local_kleio_home(path: str | None = None)[source]

Find kleio home directory.

Kleio home directory is the directory where Kleio Server finds sources and auxiliary files like structures, mappings and inferences.

It can be in the current directory, parent directory, or tests directory. It can be named “kleio-home”, “timelink-home”, or “mhk-home”.

A special case is when the current directory is “notebooks”. In this case, kleio-home is assumed to be the parent directory of “notebooks” and thus set up as the timelink-project template.

Parameters:

path (str, optional) – path to start searching from; defaults to None -> current directory.

Returns:

kleio home directory

Return type:

str

generate_token(user: str, info: TokenInfo)[source]

Generate a token for a user

Parameters:

  • user (str) – user to generate token for

  • info (TokenInfo) – token information

Returns:

kleio server API response

Return type:

dict

get_container()[source]

Get the kleio server container

Returns:

kleio server container

Return type:

docker.models.containers.Container

get_home_page(token=None) → str[source]

Get home page from Kleio Server

Parameters:

server_url (str) – url of the content in the Kleio Server

get_kleio_home()[source]

Get the kleio server home directory

Returns:

kleio server home directory

Return type:

str

get_logs(**kwargs)[source]

Get the logs of the kleio server container

Parameters:

  • **kwargs – arguments to pass to docker container logs:

  • tail (str) – number of lines to show from the end of the logs

  • since (str) – show logs since a datetime timestamp integer seconds

  • until (str) – show logs until a datetime timestamp integer seconds

  • timestamps (bool) – show timestamps

  • follow (bool) – follow log output

Returns:

kleio server logs

Return type:

str

get_report(rpt_url: str | KleioFile | None, token=None) → str | None[source]

Get report from kleio server

Parameters:

  • rpt_url (str | KleioFile) – report file url or KleioFile object

  • token (str, optional) – Kleio server token; defaults to None -> use admin token.

  • Returns – str: report content

static get_server(kleio_home: str | None = None, kleio_version='latest')[source]

Check if a kleio server is running in docker mapped to a given kleio home directory.

If yes return a KleioServer object, otherwise return None If a specific version is required, it can be specified in kleio_version

Parameters:

kleio_home (str, optional) – kleio home directory; defaults to None -> any kleio home.

Returns:

KleioServer object or None

Return type:

KleioServer or None

get_source(src: str | KleioFile, token=None) → str[source]

Get the texto of a source

Parameters:

  • src (str | KleioFile) – source file url or KleioFile object

  • token (str, optional) – Kleio server token; defaults to None -> use admin token.

Returns:

source text

Return type:

str

get_sources(path: str, recurse: str | bool = True, token=None)[source]

Get list of sources from kleio server

Parameters:

  • path (str) – path to the directory in sources

  • recurse (str) – if “yes” or True, recurse in subdirectories

Returns:

kleio server API response

Return type:

dict

get_token()[source]

Get the kleio server token

Returns:

kleio server token

Return type:

str

get_translations(path: str, recurse: str | bool = True, status: str | None = None, token: str | None = None) → list[KleioFile][source]

Get translation status from kleio server.

Returns the translation status of the kleio files in path.

Parameters:

  • path (str) – Path to the directory in sources.

  • recurse (str) – If “yes”, or True, recurse in subdirectories.

  • status (str, optional) –

    Filter by translation status. Options include:

    V = valid translations T = need translation (source more recent than translation) E = translation with errors W = translation with warnings P = translation being processed Q = file queued for translation

  • token (str, optional) – Kleio server token.

Returns:

List of KleioFile objects.

Return type:

list[KleioFile]

get_url()[source]

Get the kleio server url

Returns:

kleio server url

Return type:

str

get_url_content(server_url: str, token=None, timeout=30) → str[source]

Get content from Kleio Server

Parameters:

  • server_url (str) – url of the content in the Kleio Server

  • token (str, optional) – Kleio server token; defaults to None -> use admin token.

  • timeout (int, optional) – timeout in seconds; defaults to 30.

get_version_info() → Tuple[str, str, str] | None[source]

Fetch the home page and extract version information.

Returns:

A tuple containing the version number, build number, and date of build,

or None if the information could not be extracted.

Return type:

Optional[Tuple[str, str, str]]

host: str

kleio server host

invalidate_user(user: str)[source]

Invalidate a user

Parameters:

user (str) – user to invalidate

Returns:

kleio server API response

Return type:

dict

static is_server_running(kleio_home: str | None = None, kleio_version='latest')[source]

Check if a kleio server is running in docker mapped to a given kleio home directory.

Parameters:

  • kleio_home (str, optional) – kleio home directory; defaults to None -> any kleio home.

  • kleio_version (str, optional) – kleio server version; defaults to “latest” if not specified only checks latest

Return True of False

Parameters:

kleio_home (str, optional) – kleio home directory; defaults to None -> any kleio home.

Returns:

True if server is running, False otherwise

Return type:

bool

kleio_admin_token: str

kleio server admin token

kleio_home: str

kleio server home directory

static make_token()[source]

Get the kleio server token from environment or generate a new one if not set.

Returns:

kleio server token

Return type:

str

static start(kleio_image: str = 'timelinkserver/kleio-server', kleio_version: str | None = 'latest', kleio_home: str | None = None, kleio_admin_token: str | None = None, kleio_server_port=8088, kleio_external_port=None, kleio_server_workers=3, kleio_idle_timeout=900, kleio_conf_dir=None, kleio_source_dir=None, kleio_stru_dir=None, kleio_token_db=None, kleio_default_stru=None, kleio_debug=None, consistency: str = 'cached', port: int | None = None, update: bool = False, reuse: bool = True, stop_duplicates: bool = False)[source]

Starts a kleio server in docker

Parameters:

  • kleio_image (str) – kleio server image, defaults to “timelinkserver/kleio-server”

  • kleio_version (str, optional) – kleio-server image version, defaults to “latest”

  • kleio_home (str, optional) – kleio home directory, defaults to None -> current directory

  • kleio_token (str, optional) – kleio server admin token, defaults to None -> generate a random token

  • kleio_server_port (str, optional) – kleio server port (in the container), defaults to “8088”.

  • kleio_external_port (str, optional) – kleio server external port (in the host), defaults to “8089”.

  • kleio_server_workers (str, optional) – number kleio server workers, defaults to “3”.

  • kleio_idle_timeout (int, optional) – kleio server idle timeout, defaults to 900.

  • kleio_conf_dir (str, optional) – kleio server configuration directory, defaults to None.

  • kleio_source_dir (str, optional) – kleio server sources directory, defaults to None.

  • kleio_stru_dir (str, optional) – kleio server structures directory, defaults to None.

  • kleio_token_db (str, optional) – kleio server token database, defaults to None.

  • kleio_default_stru (str, optional) – kleio server default structure, defaults to None.

  • kleio_debug (str, optional) – kleio server debug level, defaults to None.

  • consistency (str, optional) – consistency of the volume mount, defaults to “cached”

  • port (int, optional) – port to map to 8088, defaults to None -> find a free port starting at 8088

  • update (bool, optional) – update kleio server image, defaults to False

  • reuse (bool, optional) – if True, reuse an existing kleio server container with same keio_home, defaults to True.

  • stop_duplicates (bool, optional) – if True, stop and remove duplicate containers with same kleio_home, defaults to False.

Returns:

KleioServer object

Return type:

KleioServer

stop()[source]

Stop the kleio server container

translate(path: str, recurse: str = 'yes', spawn: str = 'yes', token=None)[source]

Translate sources from kleio server

Parameters:

  • path (str) – path to the directory in sources

  • recurse (str, optional) – if “yes” recurse in subdirectories

  • spawn (str, optional) – if “yes” spawn a translation process for each file

Returns:

kleio server API response

Return type:

dict

translation_clean(path: str, recurse: str)[source]

clean translations from kleio server

Removes translation results from kleio server.

Parameters:

  • path (str) – path to the directory in sources

  • recurse (str) – if “yes” recurse in subdirectories

Returns:

kleio server API response

Return type:

dict

url: str

kleio server url

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, drop_if_exists=False, 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.

This class inherits from several mixins: - DatabaseViewsMixin: Management of database views. - DatabaseMetadataMixin: Inspection of database metadata and ORM models. - DatabaseKleioMixin: Integration with Kleio server for data imports. - DatabaseQueryMixin: High-level data access and querying methods.

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 docker container (for postgres/mysql)

Type:

Container

kserver

kleio server attached to this database, used for imports

Type:

KleioServer

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.

  • drop_if_exists (bool, optional) – if True, drop the database if it exists; defaults to False.

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

check_db()[source]

Check the database health and integrity.

This method verifies that: 1. All required ORM tables exist in the database. 2. Missing tables are created if needed. 3. For PostgreSQL, checks for and removes obsolete ‘linkstatus’ type.

If missing tables are detected, they are created automatically.

create_db()[source]

Create the database tables, views, and mappings.

This method performs the following operations: 1. Cleans caches of Mappings and ORM classes. 2. Removes dynamic tables from metadata. 3. Creates tables from ORM metadata. 4. Stamps the database with the most recent Alembic version. 5. Loads database classes and ensures all mappings are available. 6. Creates database views for common queries.

Raises:

Exception – If there is an error creating the database or updating views.

drop_db(session=None, timelink_only=False)[source]

Drop the database or only the Timelink-related tables.

Parameters:

  • session (Session, optional) – Database session to use. Defaults to None.

  • timelink_only (bool, optional) – If True, only drop Timelink-related tables and views while leaving other tables intact. If False, drops the entire database. Defaults to False.

Note

For PostgreSQL, this method attempts to terminate other active connections before dropping the database to avoid “database is being accessed by other users” errors.

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_metadata()[source]

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

has_active_connections()[source]

Check if there are active connections in the engine pool.

class timelink.api.database.TimelinkDatabaseSchema(*, db_name: str, db_type: str)[source]

Pydantic schema for TimelinkDatabase representation.

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.database.get_db_password()[source]

Get the database password from the environment or generate a new one.

Retrieves the password from the TIMELINK_DB_PASSWORD environment variable. If not set, generates a new random password and stores it in the environment.

Returns:

The database password.

Return type:

str

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

Determine the import status of Kleio files by comparing with the database.

Compares the provided Kleio files with previously imported files in the database to determine if they need to be imported, updated, or are already current. The import status is stored in each KleioFile object’s import_status attribute.

Import status values:

  • N (New): File has not been imported or translation is missing

  • U (Updated): File has been modified since last import

  • I (Imported): File successfully imported with no errors or warnings

  • E (Error): File was imported but has errors

  • W (Warning): File was imported but has warnings

Parameters:

  • db (TimelinkDatabase) – Timelink database to check against.

  • kleio_files (List[KleioFile]) – List of Kleio files to check. Can also be a single KleioFile object, which will be wrapped in a list.

  • match_path (bool, optional) – If True, matches files by full path instead of just filename. Set to True when multiple files have the same name but different paths. Defaults to False.

Returns:

The same list of Kleio files with updated import_status attributes.

Return type:

List[KleioFile]

Raises:

ValueError – If multiple files have the same name and match_path is False.

Note

The import status is returned in the KleioFile.import_status attribute. See timelink.api.models.system.KleioImportedFileSchema.import_status and timelink.kleio.import_status_enum for more details.

timelink.api.database.get_postgres_container() → Container[source]

Get the PostgreSQL Docker container object.

Returns:

The PostgreSQL container object.

Return type:

docker.models.containers.Container

Raises:

  • RuntimeError – If Docker is not running.

  • IndexError – If no PostgreSQL container is found.

timelink.api.database.get_postgres_container_pwd() → str[source]

Get the PostgreSQL container password from environment variables.

Returns:

The PostgreSQL container password.

Return type:

str

Raises:

RuntimeError – If Docker is not running.

Note

Returns None if PostgreSQL is not running or password is not found.

timelink.api.database.get_postgres_container_user() → str[source]

Get the PostgreSQL container user from environment variables.

Returns:

The PostgreSQL container user.

Return type:

str

Raises:

RuntimeError – If Docker is not running.

Note

Returns None if PostgreSQL is not running or user is not found.

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

Get the list of non-template databases from a running PostgreSQL server.

Queries the PostgreSQL system catalog to retrieve database names that: - Are not templates (NOT datistemplate) - Allow connections (datallowconn) - Are not the ‘postgres’ system database

Returns:

List of database names.

Return type:

list[str]

Raises:

RuntimeError – If Docker is not running.

Note

This function will start a PostgreSQL server if one is not already running.

Example SQL query executed:

SELECT datname
    FROM pg_database
WHERE NOT datistemplate
        AND datallowconn
        AND datname <> 'postgres';
timelink.api.database.get_postgres_url(dbname: str) → str[source]

Construct a PostgreSQL connection URL for a given database name.

Parameters:

dbname (str) – The name of the database.

Returns:

A PostgreSQL connection URL in the format:

postgresql://user:password@localhost:5432/dbname

Return type:

str

timelink.api.database.get_sqlite_databases(directory_path: str, relative_path=True) → list[str][source]

Search for and list SQLite database files in a directory.

Walks through the directory tree starting from the specified path and identifies files with .sqlite or .db extensions.

Parameters:

  • directory_path (str) – Directory path to search for SQLite databases.

  • relative_path (bool, optional) – If True, returns paths relative to the current working directory. If False, returns absolute paths. Defaults to True.

Returns:

List of SQLite database file paths.

Return type:

list[str]

timelink.api.database.get_sqlite_url(db_path: str) → str[source]

Construct an SQLAlchemy SQLite connection URL for a given file path.

Parameters:

db_path (str) – Database file path. Use “:memory:” for in-memory database.

Returns:

SQLite connection URL in SQLAlchemy format.

  • For in-memory: “sqlite:///:memory:”

  • For file: “sqlite:///path/to/file.db”

Return type:

str

timelink.api.database.import_from_xml(filespec: str | Path, session: Session, options: dict | None = None) → dict[source]

Import data from file or url into a timelink-mhk database.

The data file must be a XML file in the format exported by the kleio translator.

Parameters:

  • filespec (str,Path) – a file path, URL of a data file or path in a Kleio Server.

  • session – a database session from TimelinkDatabase.session()

  • options (dict) –

    a dictionnary with options

    • ’return_stats’: if True import stats will be returned

    • ’kleio_url’: the url of kleio server;

    • ’kleio_token’: the authorization token for the kleio server.

    • ’mode’: the mode of the import, either ‘TL’(Timelink) or ‘MHK’

  • from (If kleio_url and kleio_token are specified the data will be fetched)

  • file (a KleioServer and the filespec should contain the "xml_path" of the)

  • server (in the)

  • autentication (the kleio_token will be inserted in the header for)

Returns:

If stats is True in options a dict with statistical information will be returned.

  • ’datetime’: the time stamp of the start of the import

  • ’machine’: local machine name

  • ’file’: path to imported file or url

  • ’import_time_seconds’: elapsed time during import

  • ’entities_processed’: number of entities processed

  • ’entity_rate’: number of entities processed per second

  • ’person_rate’: number of persons (entities of class ‘person’)

  • ’nerrors’: number of errors during import

  • ’errors’: list of error messages

Examples

Returned statistical information when stats=True

{
'datetime': '2022-01-02 18:11:12',
'machine': 'joaquims-mbpr.local',
'file': 'https://...b1685.xml',
'import_time_seconds': 7.022288084030151,
'entities_processed': 747,
'entity_rate': 106.37558457603042,
'person_rate': 27.483919441999827
'nerrors': 0
'errors': []
}

TODO: should use https when the kleio_url not local.
class timelink.api.database.import_status_enum(value)[source]

An enumeration.

timelink.api.database.is_postgres_running()[source]

Check if PostgreSQL is running in a Docker container.

Returns:

True if at least one PostgreSQL container is running, False otherwise.

Return type:

bool

Note

If Docker is not running, a warning is issued and False is returned.

timelink.api.database.is_valid_postgres_db_name(db_name)[source]

Validate a PostgreSQL database name according to PostgreSQL naming rules.

Parameters:

db_name (str) – The database name to validate.

Returns:

True if the name is valid, False otherwise.

Return type:

bool

Note

A valid PostgreSQL database name must: - Be less than 64 characters long - Start with a letter or underscore - Contain only letters, digits, and underscores

timelink.api.database.random_password()[source]

Generate a random password consisting of 10 ASCII letters.

Returns:

A randomly generated 10-character string.

Return type:

str

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]

Start a PostgreSQL server in a Docker container.

If a PostgreSQL container is already running, returns the existing container. Otherwise, starts a new container with the specified configuration.

Parameters:

  • dbname (str, optional) – Initial database name. Defaults to “timelink”.

  • dbuser (str, optional) – Database user. Defaults to “timelink”.

  • dbpass (str, optional) – Database password. If None, retrieves from environment.

  • image (str, optional) – Docker image to use. Defaults to “postgres”.

  • version (str, optional) – Image version. Defaults to “latest”.

Returns:

The started or existing PostgreSQL container.

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If Docker is not running or if the server fails to start/ready.

timelink.api.database_postgres module

PostgreSQL database management for Timelink.

This module provides functions to manage a PostgreSQL server running in a Docker container. It includes utilities to check if Docker and PostgreSQL are running, retrieve PostgreSQL container details such as user credentials and connection parameters, construct PostgreSQL connection URLs, and start a PostgreSQL server in a Docker container with custom configuration.

Functions:
is_postgres_running() -> bool:

Check if PostgreSQL is running in Docker.

get_postgres_container() -> docker.models.containers.Container:

Get the PostgreSQL Docker container object.

get_postgres_container_pwd() -> str:

Get the PostgreSQL container password from environment variables.

get_postgres_container_user() -> str:

Get the PostgreSQL container user from environment variables.

get_postgres_url(dbname: str) -> str:

Construct a PostgreSQL connection URL for a given database name.

start_postgres_server(dbname: str | None = “timelink”, dbuser: str | None = “timelink”,

dbpass: str | None = None, image: str | None = “postgres”, version: str | None = “latest”) -> docker.models.containers.Container:

Start a PostgreSQL server in Docker with the specified configuration.

get_postgres_dbnames() -> list[str]:

Get the list of non-template databases from a running PostgreSQL server.

is_valid_postgres_db_name(db_name: str) -> bool:

Validate a PostgreSQL database name according to PostgreSQL naming rules.

timelink.api.database_postgres.get_postgres_container() → Container[source]

Get the PostgreSQL Docker container object.

Returns:

The PostgreSQL container object.

Return type:

docker.models.containers.Container

Raises:

  • RuntimeError – If Docker is not running.

  • IndexError – If no PostgreSQL container is found.

timelink.api.database_postgres.get_postgres_container_pwd() → str[source]

Get the PostgreSQL container password from environment variables.

Returns:

The PostgreSQL container password.

Return type:

str

Raises:

RuntimeError – If Docker is not running.

Note

Returns None if PostgreSQL is not running or password is not found.

timelink.api.database_postgres.get_postgres_container_user() → str[source]

Get the PostgreSQL container user from environment variables.

Returns:

The PostgreSQL container user.

Return type:

str

Raises:

RuntimeError – If Docker is not running.

Note

Returns None if PostgreSQL is not running or user is not found.

timelink.api.database_postgres.get_postgres_dbnames()[source]

Get the list of non-template databases from a running PostgreSQL server.

Queries the PostgreSQL system catalog to retrieve database names that: - Are not templates (NOT datistemplate) - Allow connections (datallowconn) - Are not the ‘postgres’ system database

Returns:

List of database names.

Return type:

list[str]

Raises:

RuntimeError – If Docker is not running.

Note

This function will start a PostgreSQL server if one is not already running.

Example SQL query executed:

SELECT datname
    FROM pg_database
WHERE NOT datistemplate
        AND datallowconn
        AND datname <> 'postgres';
timelink.api.database_postgres.get_postgres_url(dbname: str) → str[source]

Construct a PostgreSQL connection URL for a given database name.

Parameters:

dbname (str) – The name of the database.

Returns:

A PostgreSQL connection URL in the format:

postgresql://user:password@localhost:5432/dbname

Return type:

str

timelink.api.database_postgres.is_postgres_running()[source]

Check if PostgreSQL is running in a Docker container.

Returns:

True if at least one PostgreSQL container is running, False otherwise.

Return type:

bool

Note

If Docker is not running, a warning is issued and False is returned.

timelink.api.database_postgres.is_valid_postgres_db_name(db_name)[source]

Validate a PostgreSQL database name according to PostgreSQL naming rules.

Parameters:

db_name (str) – The database name to validate.

Returns:

True if the name is valid, False otherwise.

Return type:

bool

Note

A valid PostgreSQL database name must: - Be less than 64 characters long - Start with a letter or underscore - Contain only letters, digits, and underscores

timelink.api.database_postgres.start_postgres_server(dbname: str | None = 'timelink', dbuser: str | None = 'timelink', dbpass: str | None = None, image: str | None = 'postgres', version: str | None = 'latest')[source]

Start a PostgreSQL server in a Docker container.

If a PostgreSQL container is already running, returns the existing container. Otherwise, starts a new container with the specified configuration.

Parameters:

  • dbname (str, optional) – Initial database name. Defaults to “timelink”.

  • dbuser (str, optional) – Database user. Defaults to “timelink”.

  • dbpass (str, optional) – Database password. If None, retrieves from environment.

  • image (str, optional) – Docker image to use. Defaults to “postgres”.

  • version (str, optional) – Image version. Defaults to “latest”.

Returns:

The started or existing PostgreSQL container.

Return type:

docker.models.containers.Container

Raises:

RuntimeError – If Docker is not running or if the server fails to start/ready.

timelink.api.database_sqlite module

SQLite database utilities for Timelink.

This module provides utility functions for working with SQLite databases, including methods to search for SQLite files in a directory and construct SQLAlchemy connection URLs for SQLite files.

Functions:
get_sqlite_databases(directory_path: str, relative_path: bool = True) -> list[str]:

Search for and list SQLite databases in a specified directory.

get_sqlite_url(db_path: str) -> str:

Construct an SQLAlchemy SQLite connection URL for a given file path.

timelink.api.database_sqlite.get_sqlite_databases(directory_path: str, relative_path=True) → list[str][source]

Search for and list SQLite database files in a directory.

Walks through the directory tree starting from the specified path and identifies files with .sqlite or .db extensions.

Parameters:

  • directory_path (str) – Directory path to search for SQLite databases.

  • relative_path (bool, optional) – If True, returns paths relative to the current working directory. If False, returns absolute paths. Defaults to True.

Returns:

List of SQLite database file paths.

Return type:

list[str]

timelink.api.database_sqlite.get_sqlite_url(db_path: str) → str[source]

Construct an SQLAlchemy SQLite connection URL for a given file path.

Parameters:

db_path (str) – Database file path. Use “:memory:” for in-memory database.

Returns:

SQLite connection URL in SQLAlchemy format.

  • For in-memory: “sqlite:///:memory:”

  • For file: “sqlite:///path/to/file.db”

Return type:

str

timelink.api.database_utils module

This module provides utility functions for interacting with the Timelink database, including generating random passwords, retrieving database passwords from the environment, and determining the import status of Kleio files.

Functions:
random_password() -> str:

Generate a random password consisting of 10 ASCII letters.

get_db_password() -> str:

Retrieve the database password from the environment. If not set, generate a new one, set it in the environment, and return it.

get_import_status(db: TimelinkDatabase, kleio_files: List[KleioFile], match_path=False) -> List[KleioFile]:

Determine the import status of the provided Kleio files by comparing them with previously imported files in the database. Optionally match files by their path instead of their name.

timelink.api.database_utils.get_db_password()[source]

Get the database password from the environment or generate a new one.

Retrieves the password from the TIMELINK_DB_PASSWORD environment variable. If not set, generates a new random password and stores it in the environment.

Returns:

The database password.

Return type:

str

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

Determine the import status of Kleio files by comparing with the database.

Compares the provided Kleio files with previously imported files in the database to determine if they need to be imported, updated, or are already current. The import status is stored in each KleioFile object’s import_status attribute.

Import status values:

  • N (New): File has not been imported or translation is missing

  • U (Updated): File has been modified since last import

  • I (Imported): File successfully imported with no errors or warnings

  • E (Error): File was imported but has errors

  • W (Warning): File was imported but has warnings

Parameters:

  • db (TimelinkDatabase) – Timelink database to check against.

  • kleio_files (List[KleioFile]) – List of Kleio files to check. Can also be a single KleioFile object, which will be wrapped in a list.

  • match_path (bool, optional) – If True, matches files by full path instead of just filename. Set to True when multiple files have the same name but different paths. Defaults to False.

Returns:

The same list of Kleio files with updated import_status attributes.

Return type:

List[KleioFile]

Raises:

ValueError – If multiple files have the same name and match_path is False.

Note

The import status is returned in the KleioFile.import_status attribute. See timelink.api.models.system.KleioImportedFileSchema.import_status and timelink.kleio.import_status_enum for more details.

timelink.api.database_utils.random_password()[source]

Generate a random password consisting of 10 ASCII letters.

Returns:

A randomly generated 10-character string.

Return type:

str

timelink.api.schemas module

Pydantic Schemas for the Timelink API.

This module defines Pydantic models (schemas) used for data validation, serialization, and API communication in the Timelink information system. These models are distinct from the SQLAlchemy database models and are used to represent data for search requests, search results, and structured API responses.

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

Pydantic Schema for an Entity Attribute.

Represents a specific characteristic or property assigned to an 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.EntityAttrRelSchema(**data: Any)[source]

Pydantic Schema for an Entity with its attributes, relations, and hierarchy.

This is a comprehensive schema used to represent an entity along with all its associated data and nested structure.

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 a simplified Entity view without hierarchical links.

Used when only core entity information is needed without nested children.

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 a full Entity with its hierarchy.

id

Unique entity identifier.

Type:

str

pom_class

The POM class name of the entity.

Type:

str

inside

ID of the containing entity.

Type:

Optional[str]

the_source

Source file identifier.

Type:

Optional[str]

the_order

Order within the source.

Type:

Optional[int]

the_level

Hierarchy level.

Type:

Optional[int]

the_line

Line number in source.

Type:

Optional[int]

groupname

Kleio group name.

Type:

Optional[str]

updated

Last update timestamp.

Type:

Optional[datetime]

indexed

Last indexing timestamp.

Type:

Optional[datetime]

extra_info

Additional metadata.

Type:

Optional[dict]

contains

List of nested entities.

Type:

Optional[List[EntitySchema]]

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]

Pydantic Schema for data import statistics.

datetime

Date of the import.

Type:

date

machine

Hostname of the machine performing the import.

Type:

str

database

Name of the target database.

Type:

str

file

Path of the file being imported.

Type:

str

import_time_seconds

Total duration of the import process.

Type:

float

entities_processed

Total number of entities handled.

Type:

int

entity_rate

Entities processed per second.

Type:

float

person_rate

Persons processed per second.

Type:

float

nerrors

Total number of errors encountered.

Type:

int

errors

List of specific error messages.

Type:

List[str]

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]

Pydantic Schema for 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 a RealEntity with all its associated data.

Includes aggregated attributes and relations from linked historical 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.RealEntitySchema(**data: Any)[source]

Pydantic Schema for a RealEntity (reconciled entity) brief view.

Represents a “real world” entity that links multiple historical occurrences.

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]

Pydantic Schema for an incoming relationship including the origin name.

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]

Pydantic Schema for an outgoing relationship including the destination name.

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]

Pydantic Schema for an Entity Relationship.

Represents a directed relationship between two 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.SearchRequest(*, q: str, after: date | None = None, until: date | None = None, skip: int | None = 0, limit: int | None = 100)[source]

Pydantic Schema for a search query request.

q

The search query string.

Type:

str

after

Only return results after this date.

Type:

date, optional

until

Only return results until this date.

Type:

date, optional

skip

Number of items to skip for pagination. Defaults to 0.

Type:

int, optional

limit

Maximum number of items to return. Defaults to 100.

Type:

int, optional

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]

Pydantic Schema for a single search result entry.

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

SQLAlchemy View Support for Timelink.

This module provides utilities for defining and managing SQL views in SQLAlchemy, enabling views to be treated as tables for querying purposes. It includes DDL elements for creating and dropping views across different database dialects (PostgreSQL and SQLite) and event listeners to automate view lifecycle management.

Based on: https://github.com/sqlalchemy/sqlalchemy/wiki/Views

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

DDL element for creating a database view.

Parameters:

  • name (str) – The name of the view to create.

  • selectable (Select) – The SQLAlchemy select statement that defines the view.

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

DDL element for dropping a database view.

Parameters:

name (str) – The name of the view to drop.

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

Define a view and register its creation/drop events with MetaData.

This function creates a Table object that represents the view, allowing it to be used in SQLAlchemy queries just like a regular table. It also registers listeners to automatically create the view after tables are created and drop it before tables are dropped.

Parameters:

  • name (str) – The name of the view.

  • metadata (MetaData) – The SQLAlchemy MetaData object to register events with.

  • selectable (Select) – The select statement defining the view’s content.

Returns:

A SQLAlchemy Table object representing the view.

Return type:

Table

timelink.api.views.view_doesnt_exist(ddl, target, connection, **kw)[source]

Check if a view does not exist in the database.

timelink.api.views.view_exists(ddl, target, connection, **kw)[source]

Check if a view exists in the database.

Used as a condition for conditional DDL execution.