dtpyfw 1.19

DealerTower Python Framework: reusable building‑blocks for DealerTower services

DealerTower Python Framework (dtpyfw)

DealerTower Python Framework (dtpyfw) is a production-ready internal framework providing reusable building blocks for DealerTower microservices. It covers API development, database orchestration, caching, event streaming, object storage, task scheduling, and structured logging — all with full type safety and consistent interfaces.

PEP 561 typed package — py.typed marker is shipped so mypy/pyright narrow types in consumer services automatically.

---

Installation

Requires Python 3.13 or newer.

Base

pip install dtpyfw

Includes dtpyfw.core (env, retry, hashing, chunking, validation, …) and dtpyfw.log (structured logging).

From git tag (recommended for internal services)

dtpyfw @ git+https://github.com/datgate/dtpyfw.git@v1.0

Development

poetry install -E all

Check installed version

import dtpyfw
print(dtpyfw.__version__)  # "1.0"

Optional extras

Extras can be combined: pip install dtpyfw[api,db,redis].

Extra	Key dependencies	Install
api	FastAPI, Uvicorn, Gunicorn	pip install dtpyfw[api]
db	SQLAlchemy 2, asyncpg, psycopg2	pip install dtpyfw[db]
db-mysql	PyMySQL, aiomysql	pip install dtpyfw[db-mysql]
bucket	boto3	pip install dtpyfw[bucket]
redis	redis-py + hiredis	pip install dtpyfw[redis]
redis_streamer	redis-py	pip install dtpyfw[redis_streamer]
worker	Celery, celery-redbeat, celery_once	pip install dtpyfw[worker]
kafka	kafka-python	pip install dtpyfw[kafka]
opensearch	opensearch-py	pip install dtpyfw[opensearch]
ftp	paramiko	pip install dtpyfw[ftp]
encrypt	python-jose, passlib, bcrypt	pip install dtpyfw[encrypt]
all	Everything above	pip install dtpyfw[all]

Common profiles:

pip install dtpyfw[api,db,redis]        # API microservice
pip install dtpyfw[worker,db,redis]     # Celery worker service
pip install dtpyfw[api,db,opensearch,redis]  # Search-enabled service
pip install dtpyfw[bucket,db,ftp]       # Data processing service

---

Documentation

• CHANGELOG.md — version index linking to per-release notes in docs/changelogs/
• UPGRADE.md — migration guides linking to docs/upgrades/
• docs/ — detailed module documentation for every subpackage

---

Quick start examples

FastAPI application

from dtpyfw.api import Application
from dtpyfw.api.routes import Router, Route, RouteMethod
from dtpyfw.api.routes.authentication import Auth, AuthType
from dtpyfw.core.env import Env

gateway_auth = Auth.from_env(
    auth_type=AuthType.HEADER,
    header_key="x-gateway-key",
    env_var="gateway_key",
)

router = Router(prefix="/health", tags=["health"])
router.add_route(Route(
    method=RouteMethod.GET,
    path="/",
    endpoint=lambda: {"status": "ok"},
))

app = Application(
    title="My Microservice",
    version="1.0",
    routers=[router],
    auth=gateway_auth,
).get_app()

Database

from dtpyfw.db.config import DatabaseConfig
from dtpyfw.db.database import DatabaseInstance

db_config = DatabaseConfig.from_env()   # reads db_host, db_port, db_user, …
db = DatabaseInstance(db_config)

with db.get_db_cm_sync() as session:
    result = session.execute(select(User)).scalars().all()

Structured logging

from dtpyfw.log.config import LogConfig
from dtpyfw.log.initializer import log_initializer

log_config = LogConfig.from_env()   # reads logging_ms_url, log_level, …
log_initializer(config=log_config)

# Inside any function:
from dtpyfw.log import footprint

footprint.leave(
    log_type="info",
    controller=f"{__name__}.my_func",
    subject="Task started",
    message="Processing item.",
    payload={"item_id": item_id},
)

Redis caching

from dtpyfw.redis.config import RedisConfig
from dtpyfw.redis.connection import RedisInstance

redis = RedisInstance(RedisConfig.from_env())   # reads redis_url / redis_host, …

from dtpyfw.redis.caching import cache_function

@cache_function(redis_instance=redis, expire_time=3600)
def get_dealer(dealer_id: str) -> dict:
    ...

S3-compatible storage

from dtpyfw.bucket.bucket import Bucket

bucket = Bucket.from_env()          # reads s3_bucket_name, s3_access_key, …
bucket = Bucket.from_env("media_s3_")  # custom prefix

url = bucket.upload("path/to/file.pdf", "dealers/123/file.pdf")
bucket.download("dealers/123/file.pdf", "/tmp/file.pdf")
exists = bucket.exists("dealers/123/file.pdf")

Celery worker

from dtpyfw.worker.task import Task
from dtpyfw.worker.worker import Worker
from dtpyfw.redis.config import RedisConfig
from dtpyfw.redis.connection import RedisInstance

task = Task()
task.register("myapp.tasks.process_data", queue="default")
task.add_periodic("myapp.tasks.cleanup", crontab(hour=0, minute=0))

redis = RedisInstance(RedisConfig.from_env())
worker = Worker()
worker.set_name("my_worker").set_redis(redis).set_task(task)
celery_app = worker.get_celery()

Startup diagnostics report

Emit a structured boot-time snapshot (environment variables, dependency health, process metadata) to stdout and footprint. All dependency probes run concurrently; the function never raises regardless of which probes fail.

from dtpyfw.diagnostics import emit_startup_report

# Minimal (process info + env snapshot, no dependency probes)
emit_startup_report(service_name="my-service", process_role="api")

# With dependencies
from app.config.database import database
from app.config.redis import redis_instance, redis_queue_instance
from app.config.log import log_config

emit_startup_report(
    service_name="my-service",
    process_role="api",
    service_version="1.4.2",
    database=database,
    redis=[redis_instance, redis_queue_instance],
    log_config=log_config,
    probe_timeout_seconds=3.0,
)

# Collect without emitting
from dtpyfw.diagnostics import collect_startup_report
import json

report = collect_startup_report(service_name="my-service", process_role="script")
print(json.dumps(report, indent=2, default=str))

The returned dict always has schema_version=1 and the following top-level keys: status ("ok" / "degraded" / "down"), process, environment, logging, dependencies, extras, notes.

---

Redis Streams (event bus)

from dtpyfw.redis.config import RedisConfig
from dtpyfw.redis.connection import RedisInstance
from dtpyfw.redis_streamer.synchronize import RedisStreamer
from dtpyfw.redis_streamer.message import Message

redis = RedisInstance(RedisConfig.from_env("redis_queue_"))
streamer = RedisStreamer(redis_instance=redis, consumer_name="my_service")

streamer.register_channel("my_service_events")
streamer.subscribe("dealer_updated")
streamer.register_handler("dealer_updated", handle_dealer_updated)

# Publish
streamer.send_message("my_service_events", Message(name="event_name", body=payload))

# Consume (blocking)
streamer.persist_consume()

---

Module reference

Module	Extra	Docs
dtpyfw.core	base	docs/core/
dtpyfw.log	base	docs/log/
dtpyfw.api	api	docs/api/
dtpyfw.db	db	docs/db/
dtpyfw.bucket	bucket	docs/bucket/
dtpyfw.redis	redis	docs/redis/
dtpyfw.redis_streamer	redis_streamer	docs/redis_streamer/
dtpyfw.worker	worker	docs/worker/
dtpyfw.kafka	kafka	docs/kafka/
dtpyfw.opensearch	opensearch	docs/opensearch/
dtpyfw.ftp	ftp	docs/ftp/
dtpyfw.encrypt	encrypt	docs/encrypt/
dtpyfw.diagnostics	base	docs/diagnostics/

---

Development

# Install all dependencies
poetry install -E all

# Run tests
pytest

# Format
black .

# Lint
ruff check . --fix

# Type check
mypy dtpyfw

---

Version history

Current version: see CHANGELOG.md for full history and UPGRADE.md for migration guides.

Releases are version-driven: bump [tool.poetry] version in pyproject.toml in your PR and merge to main. CI then runs two jobs in publish.yml — release (creates the v<version> tag + GitHub Release if the version is new), then publish (OIDC-publishes to PyPI via Trusted Publishing). The flow is idempotent; re-publishing is just re-running the workflow run. See docs/RELEASING.md for the flow and the one-time PyPI setup.

---

License

DealerTower Python Framework is proprietary software. See LICENSE for complete terms and conditions.

---

Resources

• Repository: github.com/datgate/dtpyfw
• Issue Tracker: GitHub Issues