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.events |
kafka + redis_streamer |
docs/events/ |
dtpyfw.mongo |
mongo |
— |
dtpyfw.opensearch |
opensearch |
docs/opensearch/ |
dtpyfw.ftp |
ftp |
docs/ftp/ |
dtpyfw.encrypt |
encrypt |
docs/encrypt/ |
dtpyfw.mcp |
mcp |
docs/mcp/ |
dtpyfw.diagnostics |
base | — |
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
1 maintainer
Metadata
1 maintainer
Release files for dtpyfw 1.67
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | Provenance | |
|---|---|---|---|---|
| dtpyfw-1.67.tar.gz | 466.4 kB | With provenance | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| dtpyfw-1.67-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 1.0 MB
Release files / dtpyfw-1.67.tar.gz
| Download URL | dtpyfw-1.67.tar.gz |
|---|---|
| Size | 466.4 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
4bee5e50e2e2b363aef692e9af41d2952b251321d0712c715e4a70c384ce6da0
|
|
BLAKE2b-256 checksum How to use checksums |
9e1b4eb5415f2ae528686819d8359c48cc78e1c045f4d74d51c75c2dcc5042c3
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/6.1.0 CPython/3.13.13
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 14, 2026.
Transparency logRelease files / dtpyfw-1.67-py3-none-any.whl
| Download URL | dtpyfw-1.67-py3-none-any.whl |
|---|---|
| Size | 551.3 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
0f6ebf7806b3980e477c89571ae2dfc969f14ec055cdb7af5cdaf1d726f2071f
|
|
BLAKE2b-256 checksum How to use checksums |
a77725b06e8cd8673813c1d16cf7ff214d245d107a555541f0d9fb74f1fce99f
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/6.1.0 CPython/3.13.13
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 14, 2026.
Transparency logSecurity for dtpyfw 1.67
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
Full provenance
All 2 files in this release have been uploaded with provenance metadata.
Attestation for dtpyfw-1.67.tar.gz
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 14, 2026.
Transparency logA note on package security
PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.
