module-typica

Standardized Pydantic models and utilities for the things every service re-writes: database/queue connection metadata, request and response schemas, identifiers, enums, logging, and optional ready-made connectors. Define once, validate everywhere.

Install

pip install module-typica

The core package is pure Pydantic. The connectors under typica.modules need extra drivers — install only the ones you use:

pip install module-typica psycopg[binary,pool] shapely python-dateutil  # postgres
pip install module-typica pymongo            # mongo
pip install module-typica redis              # redis
pip install module-typica clickhouse-connect # clickhouse
pip install module-typica confluent-kafka    # kafka
pip install module-typica pika               # rabbitmq
pip install module-typica elasticsearch7 elasticsearch8  # elasticsearch
pip install module-typica requests requests-cache          # rest / external api

Usage

Connection metadata

Alias-rich models that accept whatever shape your config arrives in (snake_case, camelCase, common synonyms) and parse connection URIs.

from typica import DBConnectionMeta

# From a URI...
meta = DBConnectionMeta(uri="postgresql://user:pass@localhost:5432/mydb")
meta.host       # "localhost"
meta.port       # 5432
meta.database   # "mydb"

# ...or from parts, then build a URI:
meta = DBConnectionMeta(host="localhost", port=5432, username="user", password="pass", database="mydb")
meta.uri_string(base="postgresql")  # "postgresql://user:pass@localhost:5432/mydb"

# Aliases just work:
DBConnectionMeta.model_validate({"hostname": "db", "db_name": "shop"})

# Query strings, missing ports and clusters parse too:
DBConnectionMeta(uri="postgresql://h/db?sslmode=require").database  # "db"
ClusterConnectionMeta(uri="mongodb://u:p@h1:1,h2:2/?authSource=admin").database  # "admin"
RedisConnectionMeta(uri="rediss://:secret@cache:6380/2").use_ssl  # True

Available: EndpointMeta, AuthMeta, DBConnectionMeta, ClusterConnectionMeta, ESConnectionMeta, S3ConnectionMeta, RedisConnectionMeta, RMQConnectionMeta, KafkaMeta, FileConnectionMeta, HTTPConnectionMeta, DatasetMeta. parse_uri() is the shared URI parser.

ConnectionTypes enumerates one tag per meta: postgresql, mongo, clickhouse, elasticsearch, redis, s3aws, kafka, rabbitmq, file, http, mysql.

Connector contracts

Abstract bases for connectors in other repos. BaseConnector gives the context-manager shape (connect() on enter, close() on exit); the others add the CRUD / queue method set.

from typica import SQLConnector, DBConnectionMeta

class MyPgConnector(SQLConnector):
    def __init__(self, meta: DBConnectionMeta):
        super().__init__(meta)
    def connect(self, **kwargs): ...
    def close(self): ...
    def get(self, query, **kwargs): ...
    def get_all(self, query, **kwargs): ...

with MyPgConnector(meta) as pg:
    pg.get("select 1")

Available: BaseConnector, SQLConnector, NosqlConnector, QueueConnector.

Identifiers & metadata mixins

from typica import StringIdentifier, UUIDIdentifier, CreationMeta

class User(StringIdentifier, CreationMeta):
    name: str

u = User(name="ada")
u.id          # auto uuid4 string
u.created_at  # datetime (accepts epoch sec/ms or ISO strings on input)

Mongo _id variants (StringIdentifier_, UUIDIdentifier_) alias the field to _id. UpdateMeta / DeleteMeta are the updated_at / deleted_at siblings of CreationMeta. MetaCreate / MetaUpdate / MetaDelete are the camelCase epoch-millisecond variants.

Request / response schemas

from typica import PaginationSchema, FilterOpsSchema
from typica.response import ServiceResponse
from pydantic import BaseModel

PaginationSchema(page=2, size=20)
FilterOpsSchema(filter_by="age", filter_op="gte", filter_value=18)

class UserOut(BaseModel):
    id: str
    name: str

# FastAPI-style OpenAPI response maps:
responses = ServiceResponse(UserOut).get("UserGet", auth=True)
# -> {200, 400, 401, 404, 500} with generated models

Streaming payloads

from typica import StreamingStructureModel

payload = StreamingStructureModel.model_validate({
    "config": {"index": "public.users", "primary_field": "id", "method_ingestion": "upsert"},
    "data": [{"id": 1, "name": "ada"}],
})
payload.config.table_schema  # "public"
payload.to_elastic_actions()  # bulk-helper actions

Available: ConfigStreaming, StreamingStructureModel, ReportIgestorLogsMeta.

Enums

from typica.utils import DataStatus
from typica.utils.enums import Operator

Operator.gte.value        # "gte"
Operator.gte.description  # "value is greater equals to"
DataStatus.list()         # ["active", "archive", ...]

Logging

from typica.utils.log import setup_logger, CustomLogLevel

log = setup_logger("my_app", console_level=CustomLogLevel.INFO)
log.connection("connected")  # custom CONNECTION / SUCCESS levels
log.success("done")

Connectors (optional)

Thin, typed wrappers driven by the connection metadata above. Each is a context manager. Install the matching driver first (see Install).

from typica import RMQConnectionMeta
from typica.modules.rmq import RMQConnector

meta = RMQConnectionMeta(host="localhost", port=5672, exchange="events", routing_key="user.created")
with RMQConnector(meta) as rmq:
    rmq.setup_producer()
    rmq.produce({"id": 1, "name": "ada"})

from typica import ESConnectionMeta
from typica.modules.elastic import ESConnector

with ESConnector(ESConnectionMeta(host="localhost", port=9200)) as es:
    if es.is_healthy():
        for hit in es.scan("users", body={"query": {"match_all": {}}}):
            ...

from typica import HTTPConnectionMeta
from typica.modules.rest import RESTConnector

meta = HTTPConnectionMeta(base_url="https://api.example.com/v1", token="...", cache_expire=300)
with RESTConnector(meta) as api:
    api.get_json("/users", params={"page": 1})

Also available: typica.modules.psycopg.PsycopgConnector (ingestion engine), typica.modules.psycopg_alchemy.PostgreConnector (use .conn for the raw SQLAlchemy connection), typica.modules.ckafka.KafkaConnector, typica.modules.pmongo.MongoConnector, typica.modules.redis.RedisConnector / AsyncRedisConnector (async with), typica.modules.cclickhouse.CHConnector.

Contributors