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

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_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.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.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 DL execution.