Tracktolib

Tracktor Swiss-knife Utility library.

Installation

uv add tracktolib

With specific extras:

uv add tracktolib[pg,api]

Modules

logs

Utility functions to initialize logging formatting and streams.

import logging
from tracktolib.logs import init_logging

logger = logging.getLogger()
formatter, stream_handler = init_logging(logger, 'json', version='0.0.1')

pg

Async PostgreSQL helpers using asyncpg.

uv add tracktolib[pg]

import asyncpg
from tracktolib.pg import insert_one, insert_many, insert_returning, update_one, SQLExpr

conn = await asyncpg.connect('postgresql://user:pass@localhost/db')

# Insert a single row
await insert_one(conn, 'public.test', {'foo': 'bar', 'value': 1})

# Insert multiple rows
await insert_many(conn, 'public.test', [
    {'foo': 'bar', 'value': 1},
    {'foo': 'baz', 'value': 2},
])

# Insert and return generated values
new_id = await insert_returning(conn, 'public.test', {'foo': 'bar'}, returning='id')

# Upsert (insert or update on conflict)
from tracktolib.pg import Conflict
await insert_one(conn, 'public.test', {'id': 1, 'foo': 'updated'}, on_conflict=Conflict(keys=['id']))

# Update a single row
await update_one(conn, 'public.test', {'foo': 'new', 'id': 1}, keys=['id'])

# Use raw SQL expressions as values (e.g. NOW(), gen_random_uuid())
await insert_one(conn, 'public.test', {'foo': 'bar', 'created_at': SQLExpr('NOW()')})
# → INSERT INTO public.test AS t (created_at, foo) VALUES (NOW(), $1)

await update_one(conn, 'public.test', {'updated_at': SQLExpr('NOW()'), 'id': 1}, keys=['id'])
# → UPDATE public.test t SET updated_at = NOW() WHERE id = $1

pg-sync

Sync PostgreSQL helpers using psycopg (v3).

uv add tracktolib[pg-sync]

from psycopg import connect
from tracktolib.pg_sync import insert_many, fetch_one, fetch_count, fetch_all

conn = connect('postgresql://user:pass@localhost/db')

data = [
    {'foo': 'bar', 'value': 1},
    {'foo': 'baz', 'value': 2}
]
insert_many(conn, 'public.test', data)

query = 'SELECT foo from public.test order by value asc'
value = fetch_one(conn, query, required=True)  # {'foo': 'bar'}, raises if not found

assert fetch_count(conn, 'public.test') == 2

query = 'SELECT * from public.test order by value asc'
assert fetch_all(conn, query) == data

s3

Async S3 helpers using aiobotocore.

uv add tracktolib[s3]

s3-minio

S3 helpers using minio.

uv add tracktolib[s3-minio]

s3-niquests

Async S3 helpers using niquests and botocore presigned URLs.

uv add tracktolib[s3-niquests]

from tracktolib.s3.niquests import S3Session

async with S3Session(
    endpoint_url='http://localhost:9000',
    access_key='...',
    secret_key='...',
    region='us-east-1',
) as s3:
    # Object operations
    await s3.put_object('bucket', 'path/file.txt', b'content')
    content = await s3.get_object('bucket', 'path/file.txt')
    await s3.delete_object('bucket', 'path/file.txt')

    # Streaming upload (multipart for large files)
    async def data_stream():
        yield b'chunk1'
        yield b'chunk2'
    await s3.file_upload('bucket', 'large-file.bin', data_stream())

    # Bucket policy management
    policy = {'Version': '2012-10-17', 'Statement': [...]}
    await s3.put_bucket_policy('bucket', policy)
    await s3.get_bucket_policy('bucket')
    await s3.delete_bucket_policy('bucket')

    # Empty a bucket (delete all objects)
    deleted_count = await s3.empty_bucket('bucket')

    # Sync a local directory to S3 (like aws s3 sync)
    from pathlib import Path
    result = await s3.sync_directory('bucket', Path('./local'), 'remote/prefix')
    print(f"Uploaded: {result['uploaded']}, Deleted: {result['deleted']}, Skipped: {result['skipped']}")

    # With delete flag to remove remote files not present locally
    result = await s3.sync_directory('bucket', Path('./local'), 'remote/prefix', delete=True)

api

FastAPI utilities using fastapi and pydantic.

uv add tracktolib[api]

gh

GitHub API helpers using niquests.

uv add tracktolib[gh]

from tracktolib.gh import GitHubClient

async with GitHubClient() as gh:  # Uses GITHUB_TOKEN env var
    # Issue comments
    comments = await gh.get_issue_comments("owner/repo", 123)
    await gh.create_issue_comment("owner/repo", 123, "Hello!")
    await gh.delete_comments_with_marker("owner/repo", 123, "<!-- bot -->")

    # Labels
    labels = await gh.get_issue_labels("owner/repo", 123)
    await gh.add_labels("owner/repo", 123, ["bug", "priority"])
    await gh.remove_label("owner/repo", 123, "wontfix")

    # Pull requests
    pr = await gh.get_pull_request("owner/repo", 42)
    diff = await gh.get_pull_request_diff("owner/repo", 42)
    await gh.create_pull_request_review(
        "owner/repo", 42,
        body="Found a few things",
        event="REQUEST_CHANGES",
        comments=[{"path": "app/main.py", "line": 12, "side": "RIGHT", "body": "This can be None here"}],
    )

    # Permissions
    permission = await gh.get_collaborator_permission("owner/repo", "octocat")  # admin, write, read or none

    # Deployments
    deploys = await gh.get_deployments("owner/repo", environment="production")
    await gh.mark_deployment_inactive("owner/repo", "preview-123")

cf

Cloudflare DNS API helpers using niquests.

uv add tracktolib[cf]

from tracktolib.cf import CloudflareDNSClient

async with CloudflareDNSClient() as cf:  # Uses CLOUDFLARE_API_TOKEN and CLOUDFLARE_ZONE_ID env vars
    # Get a DNS record
    record = await cf.get_dns_record("app.example.com", "CNAME")

    # Create a DNS record
    record = await cf.create_dns_record(
        "app.example.com",
        "target.example.com",
        record_type="CNAME",
        ttl=60,
        proxied=True,
    )

    # Delete by ID or name
    await cf.delete_dns_record(record["id"])
    await cf.delete_dns_record_by_name("app.example.com", "CNAME")

    # Check existence
    exists = await cf.dns_record_exists("app.example.com")

tests

Testing utilities using deepdiff.

uv add tracktolib[tests]

Metadata

Release files for tracktolib 1.1.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for tracktolib 1.1.0
File	Size	Uploaded
tracktolib-1.1.0.tar.gz	36.8 kB	Sep 23, 2026	Details

Built distribution (wheel)

Table of built distributions (wheels) for tracktolib 1.1.0
File	Interpreter	ABI	Platform	Reset
tracktolib-1.1.0-py3-none-any.whl 43.7 kB Sep 23, 2026	Python 3	none	any	Details

Total release size: 80.5 kB

Release files / tracktolib-1.1.0.tar.gz

Download URL	tracktolib-1.1.0.tar.gz
Size	36.8 kB
Tags	Source
SHA-256 checksum How to use checksums	`2f088366cfc48f91291b375c47813b64c73378ca3b31871d60f9a5dd87f56cd6`
BLAKE2b-256 checksum How to use checksums	`011467756657c95f11eb72764b574996208bd08e3fa6ce3f29a79ecde17f4f53`
Upload date	Sep 23, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`uv/0.12.18 {"installer":{"name":"uv","version":"0.12.18","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`

Release files / tracktolib-1.1.0-py3-none-any.whl

Download URL	tracktolib-1.1.0-py3-none-any.whl
Size	43.7 kB
Tags	Python 3
SHA-256 checksum How to use checksums	`a16cf05bbbd681df852e4f23d7f45689c74d74a521bf7641117a3a645789fdca`
BLAKE2b-256 checksum How to use checksums	`05bd0eb112d9d7d68f8cffa1d31555229c2f50977bd43f92f9a14278fde850b2`
Upload date	Sep 23, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	Yes
Uploaded via	`uv/0.12.18 {"installer":{"name":"uv","version":"0.12.18","subcommand":["publish"]},"python":null,"implementation":{"name":null,"version":null},"distro":{"name":"Ubuntu","version":"24.04","id":"noble","libc":null},"system":{"name":null,"release":null},"cpu":null,"openssl_version":null,"setuptools_version":null,"rustc_version":null,"ci":true}`