lokit.database

A complete PostgreSQL translation-memory layer for pooled connections, bounded ingestion, matching, retrieval, and streaming.

The database API is more than matching

Task	Synchronous API	Asynchronous API
Connect	`connect_sync()`	`await connect()`
Initialize schema	`setup_sync()`	`await setup()`
Ingest a document	`load_sync()`	`await load()`
Match one or many	`match_sync()`, `match_batch_sync()`	`await match()`, `await match_batch()`
Fetch one unit	`unit_sync()`	`await unit()`
Build a document	`to_document_sync()`	`await to_document()`
Build a multilingual document	`to_multilingual_document_sync()`	`await to_multilingual_document()`
Stream stored units	—	`stream()` returns an async iterator
Close pools	`close_sync()`	`await close()`

Configure writer and reader pools

Pass one URI for a shared read/write pool or add reader_uri for a replica or independently scaled reader pool. pool_size, min_size, SSL, timeouts, static passwords, and password factories are explicit. Pipeline mode is enabled by default and requires PostgreSQL 14 or newer; pass pipeline=False for an older server.

connect_tm.py

import lokit
tm = lokit.database.connect_sync(    "postgresql://writer@primary/lokit",    reader_uri="postgresql://reader@replica/lokit",    pool_size=8,    min_size=2,    ssl=True,    pipeline=True,)try:    tm.setup_sync(partitioned=True)finally:    tm.close_sync()

Choose who owns the schema

For a standalone database, setup() or setup_sync() creates pg_trgm, pgcrypto, the metadata table, the translation-memory tables, indexes, and the selected partitioning layout. Applications that already own migrations should apply Lokit's ordered SQL through Alembic instead. Partitioning is fixed when the schema is created; a later setup call with a different value raises rather than silently rebuilding data. See Schema & Alembic.

Load materialized or streaming documents

load() accepts BaseStructure or StreamingStructure. It lazily serializes every locale target, batches rows, deduplicates within each batch, stages units, tags, segment parts, and comments with PostgreSQL COPY, then maps or upserts them transactionally. project and domain can classify a whole load; LoadStats reports units read, units written, and elapsed seconds.

load_tm.py

import lokit
tm = lokit.database.connect_sync(dsn)try:    document = lokit.stream.tmx(        "memory.tmx",        target_language="fr",    )    stats = tm.load_sync(        document,        batch_size=5_000,        project="storefront",        domain="checkout",        progress=False,    )    print(stats.units_read, stats.units_written, stats.seconds)finally:    tm.close_sync()

Match exact, contextual, tagged, or fuzzy candidates

Matches are always scoped by source and target locale. Source and context hashes accelerate exact and in-context candidates; pg_trgm supplies similarity candidates. Previous and next source context, source tags, an explicit tag signature, and require_tags refine scoring. limit must be positive and threshold must be between 0 and 1.

match_tm.py

matches = tm.match_sync(    source="Complete your purchase",    source_locale="en-US",    target_locale="fr-FR",    previous_source="Your basket",    next_source="Order confirmation",    require_tags=True,    limit=5,    threshold=0.72,)

Batch matching has a typed input shape

MatchInput is a public TypedDict. A batch reuses database connections and, when pipeline mode is enabled, pipelines compatible work. Results preserve input order as one result list per query.

match_batch.py

from lokit.database import MatchInput
queries: list[MatchInput] = [    {        "source": "Hello",        "source_locale": "en-US",        "target_locale": "fr-FR",    },    {        "source": "Place order",        "source_locale": "en-US",        "target_locale": "fr-FR",        "previous_source": "Your basket",    },]
results = tm.match_batch_sync(    queries,    limit=5,    threshold=0.7,    progress=False,)

Retrieve units and reconstruct documents

unit() reconstructs one Data value and raises KeyError when the key is absent. to_document() returns a bilingual BaseStructure; to_multilingual_document() groups available locales into TargetData entries. Set include_tags=False when inline tags and segment parts are not needed.

retrieve.py

unit = tm.unit_sync(    "checkout.submit",    source_locale="en-US",    target_locale="fr-FR",)
document = tm.to_document_sync(    source_locale="en-US",    target_locale="fr-FR",)
multilingual = tm.to_multilingual_document_sync(    source_locale="en-US",    target_locales=("fr-FR", "de-DE"),)

Stream stored units without materializing a document

stream() returns a TranslationMemoryStream, an asynchronous context manager and iterator. It uses a server-side cursor and bounded fetch batches, then retrieves child rows for each batch before yielding reconstructed (unit_key, Data) pairs.

stream_database.py

import lokit
tm = await lokit.database.connect(dsn)async with tm:    async with tm.stream(        source_locale="en-US",        target_locale="fr-FR",        batch_size=1_000,    ) as units:        async for unit_key, unit in units:            print(unit_key, unit.target)

Use the async lifecycle for services

TranslationMemory is an async context manager, making it the natural choice for long-running services and request workers. The _sync methods are convenience bridges for scripts and synchronous applications. Both surfaces share the same pools, schema, serialization, matching, and validation behavior.