module-typica 0.2.14

Standard Pydantic usages & utilities

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

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"})

Available: EndpointMeta, AuthMeta, DBConnectionMeta, ClusterConnectionMeta, ESConnectionMeta, S3ConnectionMeta, RedisConnectionMeta, RMQConnectionMeta, KafkaMeta, FileConnectionMeta, DatasetMeta.

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.

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

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": {}}}):
            ...

Also available: typica.modules.psycopg.PsycopgConnector (ingestion engine), typica.modules.ckafka.KafkaConnector, typica.modules.pmongo.MongoConnector, typica.modules.redis.RedisConnector / AsyncRedisConnector, typica.modules.cclickhouse.CHConnector.

Contributors