faze-utils 0.1.1

Centralized, reusable utilities for Python applications (Python counterpart of go-utils).

faze-utils

Centralized, reusable utilities for Python applications — the Python counterpart of the internal go-utils module.

Supports Python 3.9+. Core install has zero runtime dependencies; every backend integration (Redis, MongoDB, PostgreSQL, Aerospike, Pub/Sub, GCS, S3) imports its driver lazily and is enabled via an extra.

Installation

pip install faze-utils                       # core: config, logs, errors, validation, ...
pip install 'faze-utils[redis]'              # + cache
pip install 'faze-utils[mongo,postgres]'     # + db backends
pip install 'faze-utils[pubsub]'             # + pubsub and queue
pip install 'faze-utils[gcs,s3]'             # + upload packages
pip install 'faze-utils[all]'                # everything (except aerospike)

Package structure

Package	Ported from	What it does
faze_utils.config	config/	Viper-equivalent config: ENV_FILE/CONFIG JSON loading, case-insensitive dot-notation typed getters, is_set, internal-service URL registry, proxy URLs
faze_utils.logs	logs/	stdlib-logging setup: JSON/INFO in prod, readable/DEBUG otherwise, IST timestamps, OTel trace-id correlation via with_context()
faze_utils.cache	cache/	Full Redis wrapper (Cache): strings/JSON, counters, TTLs, hashes, lists, sets, NX locks, pattern scan/delete, pipelines
faze_utils.db	db/	Connection factories: init_mongo_db, init_postgres_db/close_postgres_db, init_aerospike_db/close_aerospike_db
faze_utils.pubsub	pubsub/	GCP Pub/Sub publisher/subscribers, queueName envelope compatibility, OTel propagation, idempotent topology bootstrap with DLQ support
faze_utils.queue	queue/	Simple Pub/Sub string-message client with auto-ack-on-success consumer
faze_utils.request	request/	camelCase wire error codes, RequestError, {"status": ...} envelopes
faze_utils.response	response/	ServiceError + ErrorCode, HTTP status mapping, {"success": ...} envelopes
faze_utils.slack	utils/slack.go	send_slack_alert via chat.postMessage (stdlib HTTP)
faze_utils.upload_files	utils/upload.files.go	Upload validation, safe unique object naming, GCS upload
faze_utils.upload_s3_compat	utils/upload_s3_compat.go	GCSUploader for any S3-compatible endpoint: unique/conflict-check/overwrite uploads, raw-bytes upload
faze_utils.validation	validation/	Declarative field rules (required, lengths, bounds, email, pattern, custom, nested) raising VALIDATION_FAILED ServiceErrors

Configuration example

from faze_utils.config import Config

# From the CONFIG env var or ./env.<ENV_FILE>.json:
cfg = Config.load()

# Or construct directly (tests, scripts):
cfg = Config({
    "environment": "prod",
    "redis": {"address": "localhost:6379", "db": 0},
    "postgres": {"user": "svc", "password": "...", "host": "db", "port": 5432,
                  "dbname": "app", "sslmode": "require"},
    "pubSub": {"serviceAccount": {...}, "topicSubscriptions": {
        "orders": [{"name": "orders-sub", "maxDeliveryAttempts": 5}],
    }},
})

cfg.get_string("redis.address")   # "localhost:6379"
cfg.get_int("postgres.port")      # 5432
cfg.is_set("mongodb.minPoolSize") # distinguishes unset from explicit 0
cfg.get_service_url("walletService")

Usage examples

# logs
from faze_utils import logs
logger = logs.new_logger("prod")          # JSON lines, INFO+
logs.with_context().info("handling job")  # + trace_id/span_id when OTel active

# cache (pip install 'faze-utils[redis]')
from faze_utils.cache import Cache, KeyNotFoundError
cache = Cache.from_config(cfg)
cache.set_json("user:1", {"name": "a"}, expiration=300)
try:
    value = cache.get("user:1")
except KeyNotFoundError:
    ...

# db (pip install 'faze-utils[mongo]' / '[postgres]' / '[aerospike]')
from faze_utils.db import init_mongo_db, init_postgres_db, close_postgres_db
mongo = init_mongo_db(cfg)
pool = init_postgres_db(cfg)
with pool.connection() as conn:
    conn.execute("SELECT 1")
close_postgres_db(pool)

# pubsub (pip install 'faze-utils[pubsub]')
from faze_utils.pubsub import PubSub
ps = PubSub.from_config(cfg)
ps.ensure_topology_from_config(cfg)   # idempotent create-only bootstrap
ps.publish("orders", {"orderId": 42}, attrs={"region": "us"})
ps.start_subscribers({"orders": lambda msg: (process(msg), msg.ack())},
                     topics=["orders"])

# queue
from faze_utils.queue import PubSubClient
q = PubSubClient.from_config(cfg)
q.send_message_to_queue("jobs", "payload")
q.receive_message("jobs", process_func=handle)   # acks on success

# request / response envelopes
from faze_utils import request as req
from faze_utils.response import not_found, success_response
err = req.not_found_error("player not found")
body, status = req.error_envelope(err), err.http_status

# validation
from faze_utils import validation as v
v.validate_or_raise(payload, {
    "name": [v.required, v.min_length(2)],
    "email": [v.required, v.email],
    "team.name": [v.required],
})  # raises ServiceError(VALIDATION_FAILED) listing every failed field

# slack
from faze_utils.slack import send_slack_alert
send_slack_alert("#alerts", "deploy finished", "deploy-bot", icon_emoji=":rocket:")

# uploads (pip install 'faze-utils[gcs]' / '[s3]')
from faze_utils.upload_files import upload_file_to_gcp
from faze_utils.upload_s3_compat import GCSUploader
resp = upload_file_to_gcp(fileobj, "reports/q3", "media-bucket",
                          original_filename="q3.pdf")
uploader = GCSUploader.from_env("media-bucket")
resp = uploader.upload_file_with_conflict_check(fileobj, "docs/report.txt")

Development

python3 -m venv .venv && source .venv/bin/activate
pip install -e '.[dev]'        # install with dev dependencies

pytest                         # run tests
pytest --cov                   # tests with coverage
ruff format src tests          # format
ruff check src tests           # lint
mypy                           # type-check

python -m build                # build sdist + wheel into dist/
twine check dist/*             # validate the distributions

Tests use fakeredis and mocks — no real Redis, databases, brokers, Slack, or cloud accounts are contacted.

Publishing

Confirm the package name, version (pyproject.toml), and author details first. You need API tokens for TestPyPI and PyPI.

# 1. Clean build + validate
rm -rf dist/ && python -m build && twine check dist/*

# 2. Publish to TestPyPI first
twine upload --repository testpypi dist/*
# username: __token__, password: your TestPyPI API token

# 3. Verify the TestPyPI install in a scratch environment
pip install --index-url https://test.pypi.org/simple/ \
    --extra-index-url https://pypi.org/simple/ faze-utils

# 4. Publish to PyPI
twine upload dist/*

Security guidance

• Never commit credentials. All secrets flow in via the CONFIG env var, env files excluded from VCS, or dedicated env vars (SLACK_TOKEN, GCS_ACCESS_KEY_ID, GCS_SECRET_ACCESS_KEY).
• Config/DB error messages and logs never echo secret values (the postgres factory deliberately does not log the DSN, unlike the Go original).
• Store PyPI tokens in ~/.pypirc (chmod 600) or TWINE_USERNAME=__token__ / TWINE_PASSWORD env vars.
• upload_files rejects path-traversal object names (.., absolute paths, backslashes).

Behavioral differences from the Go implementation

• request: too_many_requests_error returns HTTP 429; the Go version returned 404 (a bug).
• cache: no context.Context parameters (sync redis-py); expirations accept seconds or timedelta; misses raise KeyNotFoundError instead of returning an error string.
• db/postgres: connection URL (with password) is not logged.
• db/aerospike: Go's ConnectionQueueSize has no direct equivalent in the Python client and is not applied; TLS "skip verify" is likewise driver-managed.
• logs: with_context() takes no arguments — the active OTel span is ambient in Python.
• pubsub/queue: subscriber topics are passed explicitly (Go read pubSub.subscribers config inside the function); OTel tracing degrades to a no-op when opentelemetry isn't installed; PublishV2 is folded into publish(..., ordering_key=...).
• upload_files: takes file objects + field values instead of *http.Request; adds path-traversal protection.
• upload_s3_compat: NewGCSUploaderFromViper became GCSUploader.from_config (reads the same gcs.options keys).
• response: gRPC status converters were not ported (would force a grpcio dependency); HTTP mapping is complete.
• config: formatEnvKeys (dotenv-style key rewriting) was not ported — nothing in the migrated packages uses it.

License

MIT