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.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
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Filter files by name, interpreter, ABI, and platform.
If you're not sure about the file name format, learn more about wheel file names.
Copy a direct link to the current filters
File details
Details for the file dtpyfw-1.35.tar.gz.
File metadata
- Download URL: dtpyfw-1.35.tar.gz
- Upload date:
- Size: 235.4 kB
- Tags: Source
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/6.1.0 CPython/3.13.13
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
53171f0b20c692f285e189550d32a1044fc39ada5d9fa4531192576949e27c7b
|
|
| MD5 |
d4b5d60455764b4ad4a1959066fcaf50
|
|
| BLAKE2b-256 |
d086a3097f7182dc68a3b139eaa1279385bfc688a30e834b9dd54f095babec0b
|
Provenance
The following attestation bundles were made for dtpyfw-1.35.tar.gz:
Publisher:
publish.yml on datgate/dtpyfw
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
dtpyfw-1.35.tar.gz -
Subject digest:
53171f0b20c692f285e189550d32a1044fc39ada5d9fa4531192576949e27c7b - Sigstore transparency entry: 2782722358
- Sigstore integration time:
-
Permalink:
datgate/dtpyfw@79b69a6ec6cc6bb4416267388aad708524abfda8 -
Branch / Tag:
refs/heads/main - Owner: https://github.com/datgate
-
Access:
private
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish.yml@79b69a6ec6cc6bb4416267388aad708524abfda8 -
Trigger Event:
push
-
Statement type:
File details
Details for the file dtpyfw-1.35-py3-none-any.whl.
File metadata
- Download URL: dtpyfw-1.35-py3-none-any.whl
- Upload date:
- Size: 293.8 kB
- Tags: Python 3
- Uploaded using Trusted Publishing? Yes
- Uploaded via:
twine/6.1.0 CPython/3.13.13
File hashes
| Algorithm | Hash digest | |
|---|---|---|
| SHA256 |
f7df38803febd08e1c6b2a07721c3115424962de38095c96712593e67c0d4131
|
|
| MD5 |
48be9b76f3b84d65594576fb5676a683
|
|
| BLAKE2b-256 |
b475a8aeefb9fd6db1831f7f59566eb521ed549262f06e2b052dd0cc3abb0717
|
Provenance
The following attestation bundles were made for dtpyfw-1.35-py3-none-any.whl:
Publisher:
publish.yml on datgate/dtpyfw
-
Statement:
-
Statement type:
https://in-toto.io/Statement/v1 -
Predicate type:
https://docs.pypi.org/attestations/publish/v1 -
Subject name:
dtpyfw-1.35-py3-none-any.whl -
Subject digest:
f7df38803febd08e1c6b2a07721c3115424962de38095c96712593e67c0d4131 - Sigstore transparency entry: 2782723003
- Sigstore integration time:
-
Permalink:
datgate/dtpyfw@79b69a6ec6cc6bb4416267388aad708524abfda8 -
Branch / Tag:
refs/heads/main - Owner: https://github.com/datgate
-
Access:
private
-
Token Issuer:
https://token.actions.githubusercontent.com -
Runner Environment:
github-hosted -
Publication workflow:
publish.yml@79b69a6ec6cc6bb4416267388aad708524abfda8 -
Trigger Event:
push
-
Statement type:
