paulpas Data-Layer Deduplication
Identifies and eliminates data-layer duplication — schema definitions, transformation logic, API response shapes, query patterns, and configuration schemas — by extracting canonical sources. This skill teaches how to find duplicated data definitions across tables, models, pipelines, and endpoints, then consolidate them into single authoritative sources with zero breaking changes during migration.
TL;DR Checklist
- Inventory every table, model, endpoint schema, and query that defines the same logical field
- Mark each field's claimed owner — only one definition may be canonical per domain
- Require ≥3 independent definitions before flagging as duplication candidate
- Extract shared schemas using backward-compatible dual-write (write-old + write-new) during transition
- Validate referential integrity after consolidation — no orphaned FK references or null cascades
- Update all API contract consumers before removing duplicate endpoint definitions
When to Use
Use this skill when:
- You discover the same fields (e.g.,
address,phone_number,email) defined identically in 3+ tables, models, or serializers with drift over time - ETL pipelines contain copy-pasted transformation logic for timestamp normalization, currency conversion, or country code mapping across ≥2 ingestion scripts
- API endpoints return differently-shaped pagination envelopes, error objects, or metadata fields making client integration brittle
- Report generation functions each reimplement the same complex JOIN/WHERE patterns with slight variations
- Multiple services independently validate identical configuration schemas (e.g., database URLs, feature flags) from scratch
- Data validation rules for emails, phone numbers, or identifiers appear in model layers, serializers, CLI handlers, and test fixtures — all with different regex or logic
When NOT to Use
Avoid this skill for:
- Code-level duplication of functions or classes (use
deduplication-patternsinstead) - Single-table normalization where fields belong within one entity's scope — that is database normalization, not deduplication
- One-off scripts or throwaway data processing pipelines where overhead outweighs benefit
- Scenarios where independent definitions serve genuinely different purposes (e.g., a billing address vs. shipping address are the same schema but separate instances)
Core Workflow
Map Data Ownership — Inventory every table, model class, endpoint response schema, and query template that defines logical fields. For each field, record its location, definition source, and claimed owner. Checkpoint: Fields appearing in ≥3 places with independently written definitions are your duplication candidates.
Identify Canonical Candidates — Among duplicated fields, select which definition becomes canonical. Prefer the definition used by the most consumers or the one maintained by the owning domain team. If no clear owner exists, choose the simplest, most complete definition. Checkpoint: Document the chosen canonical source and update your data dictionary with a single authoritative reference for each deduplicated field.
Assess Change Frequency — Only consolidate definitions that have changed across ≥2 release cycles or are modified by >1 team iteration. Static, one-time schemas can remain duplicated without penalty. High-frequency-change fields create the strongest case for centralization. Checkpoint: If a field has been stable since initial deployment and has fewer than 2 consumers, defer consolidation — the coupling cost outweighs the benefit.
Choose Consolidation Strategy — Based on access patterns, pick the right extraction approach:
- Schema duplication → Extract to shared type/class or reference entity with foreign key
- Transformation duplication → Create composable transform pipeline with step functions
- API response duplication → Build shared response envelope types and serializer base classes
- Query duplication → Parameterize query builder functions that assemble common JOIN/WHERE clauses
- Configuration duplication → Define a base config schema with per-service extension hooks
Checkpoint: Every chosen strategy must support backward-compatible dual-write during the transition period.
Execute Migration with Dual-Write — For each consolidation: write to both old and new locations simultaneously, verify data parity, then remove readers from old paths one domain at a time. Run referential integrity checks (foreign key constraints, nullability, type compatibility) after every consumer switch. Checkpoint: No consumer should break mid-migration; run integration tests against the dual-write state before removing any old reader.
Implementation Patterns
Pattern 1: Schema Centralization
Define data models once as shared types or reference entities. All consumers import and reference the canonical definition rather than redefining identical fields locally. This eliminates schema drift where the same logical field evolves differently in separate services over time.
BAD — Same user address duplicated across three tables/models:
# ❌ BAD: users.py — User model with inline address fields
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
address_line_1: Mapped[str] # duplicated
address_line_2: Mapped[str | None] # duplicated
city: Mapped[str] # duplicated
state_province: Mapped[str] # duplicated
postal_code: Mapped[str] # duplicated
country_code: Mapped[str] # duplicated
# ❌ BAD: orders.py — Order model repeats the same address fields
class Order(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
shipped_to_line_1: Mapped[str] # duplicated with different naming
shipped_to_line_2: Mapped[str | None] # duplicated with different naming
shipped_to_city: Mapped[str] # duplicated with different naming
shipped_to_state: Mapped[str] # duplicated with different naming
shipped_to_zip: Mapped[str] # duplicated with different naming
shipped_to_country: Mapped[str] # duplicated with different naming
# ❌ BAD: shipments.py — Shipment model yet again
class Shipment(Base):
__tablename__ = "shipments"
id: Mapped[int] = mapped_column(primary_key=True)
order_id: Mapped[int] = mapped_column(ForeignKey("orders.id"))
delivery_address_1: Mapped[str] # duplicated with yet another naming
delivery_address_2: Mapped[str | None]
delivery_city: Mapped[str]
delivery_state: Mapped[str]
delivery_postal: Mapped[str]
delivery_country: Mapped[str]
GOOD — Address as a shared type, referenced everywhere:
# ✅ GOOD: models/address.py — Single canonical address definition
from dataclasses import dataclass, field
from typing import ClassVar
@dataclass(frozen=True)
class Address:
"""Canonical address model used across all domain entities."""
line_1: str
line_2: str | None = None
city: str
state_province: str
postal_code: str
country_code: str # ISO 3166-1 alpha-2, e.g., "US", "GB"
_country_codes: ClassVar[set[str]] = {
"US", "CA", "GB", "DE", "FR", "JP", "AU", "IN", "BR", "MX"
}
def validate(self) -> list[str]:
"""Return validation errors. Empty list means valid."""
errors: list[str] = []
if not self.line_1.strip():
errors.append("line_1 is required")
if self.country_code and self.country_code.upper() not in self._country_codes:
errors.append(f"Invalid country code: {self.country_code}")
return errors
# ✅ GOOD: models/user.py — User references Address via FK or nested type
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str]
address_id: Mapped[int | None] = mapped_column(
ForeignKey("addresses.id"), nullable=True
)
# ORM relationship to the canonical Address entity
address: Mapped["Address"] = relationship(back_populates="user")
class Order(Base):
__tablename__ = "orders"
id: Mapped[int] = mapped_column(primary_key=True)
user_id: Mapped[int] = mapped_column(ForeignKey("users.id"))
billing_address_id: Mapped[int] = mapped_column(ForeignKey("addresses.id"))
shipping_address_id: Mapped[int] = mapped_column(ForeignKey("addresses.id"))
billing_address: Mapped["Address"] = relationship(
foreign_keys=[billing_address_id]
)
shipping_address: Mapped["Address"] = relationship(
foreign_keys=[shipping_address_id]
)
# ✅ GOOD: SQL migration — canonical address table created once
"""
CREATE TABLE addresses (
id SERIAL PRIMARY KEY,
line_1 VARCHAR(255) NOT NULL,
line_2 VARCHAR(255),
city VARCHAR(100) NOT NULL,
state_province VARCHAR(100) NOT NULL,
postal_code VARCHAR(20) NOT NULL,
country_code CHAR(2) NOT NULL,
created_at TIMESTAMPTZ DEFAULT now()
);
ALTER TABLE users ADD COLUMN address_id REFERENCES addresses(id);
ALTER TABLE orders ADD COLUMN billing_address_id REFERENCES addresses(id);
ALTER TABLE orders ADD COLUMN shipping_address_id REFERENCES addresses(id);
"""
When to apply: When ≥2 entities need the same logical fields, and those fields share validation rules. The canonical entity becomes the single source of truth for address format, validation, and storage.
Pattern 2: Transformation Pipeline Abstraction
Extract common ETL/ELT transformation steps into composable step functions. Instead of duplicating timestamp normalization, currency conversion, or locale formatting across every ingestion script, define a pipeline framework where each step is a reusable transform function applied in a fixed sequence.
BAD — Timestamp normalization repeated across five ingestion scripts:
# ❌ BAD: ingest_users.py — ad-hoc timestamp parsing
import re
def parse_user_row(row: dict) -> dict:
"""Ingest a single user row with inline timestamp parsing."""
raw = row["created_at"]
# Pattern 1: ISO 8601 with timezone
if "T" in raw and "+" in raw:
dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
# Pattern 2: US date format (inconsistent!)
elif re.match(r"\d{2}/\d{2}/\d{4}", raw):
dt = datetime.strptime(raw, "%m/%d/%Y").replace(
tzinfo=timezone.utc
)
# Pattern 3: Unix timestamp
elif raw.isdigit():
dt = datetime.fromtimestamp(int(raw), tz=timezone.utc)
else:
raise ValueError(f"Unparseable date: {raw}")
row["created_at"] = dt.isoformat()
return row
# ❌ BAD: ingest_orders.py — same logic, slightly different implementation
import re
def parse_order_row(row: dict) -> dict:
"""Ingest a single order row with inline timestamp parsing."""
raw = row["order_date"]
if "T" in raw and "+" in raw:
dt = datetime.fromisoformat(raw.replace("Z", "+00:00"))
elif re.match(r"\d{2}-\d{2}-\d{4}", raw): # different format!
dt = datetime.strptime(raw, "%m-%d-%Y").replace(
tzinfo=timezone.utc
)
elif raw.isdigit():
dt = datetime.fromtimestamp(int(raw), tz=timezone.utc)
else:
raise ValueError(f"Unparseable date: {raw}")
row["order_date"] = dt # inconsistent return type — some scripts return string
return row
# ❌ BAD: ingest_products.py — third implementation with yet another bug
def parse_product_row(row: dict) -> dict:
raw = row["created_at"]
try:
dt = datetime.fromisoformat(raw) # no timezone handling, crashes on Z suffix
except ValueError:
dt = datetime.now() # silent fallback to now() — data loss!
return row
GOOD — Centralized transform pipeline with composable step functions:
# ✅ GOOD: transforms/pipeline.py — Single canonical transformation framework
from collections.abc import Callable
from dataclasses import dataclass, field
from datetime import datetime, timezone
@dataclass(frozen=True)
class TransformStep:
"""A named transformation function that converts input to output."""
name: str
func: Callable[[dict], dict]
def __call__(self, record: dict) -> dict:
return self.func(record)
@dataclass
class TransformPipeline:
"""Composable ETL pipeline. Steps execute sequentially in order."""
steps: list[TransformStep] = field(default_factory=list)
def add(self, step: TransformStep) -> "TransformPipeline":
"""Chain a new step onto the pipeline."""
self.steps.append(step)
return self
def run(self, record: dict) -> dict:
"""Execute all steps sequentially on the input record."""
for step in self.steps:
try:
record = step(record)
except Exception as exc:
raise TransformError(
f"Step '{step.name}' failed on record: {exc}"
) from exc
return record
def normalize_timestamp(record: dict, target_field: str = "created_at") -> dict:
"""Normalize a timestamp field to UTC ISO 8601 string.
Handles ISO 8601 with timezone, US date format (MM/DD/YYYY),
Unix timestamps, and fails on unrecognized formats.
Args:
record: Input data row containing the timestamp field.
target_field: Key name of the timestamp in the record dict.
Returns:
Record dict with the timestamp converted to ISO 8601 UTC string.
Raises:
ValueError: If the timestamp cannot be parsed by any known format.
"""
raw = record.get(target_field)
if raw is None:
return record
dt: datetime | None = None
# Format 1: ISO 8601 (with or without timezone offset)
if isinstance(raw, str) and "T" in raw:
normalized = raw.replace("Z", "+00:00")
dt = datetime.fromisoformat(normalized)
# Format 2: US date MM/DD/YYYY
elif isinstance(raw, str) and re.match(r"\d{2}/\d{2}/\d{4}", raw):
dt = datetime.strptime(raw, "%m/%d/%Y").replace(tzinfo=timezone.utc)
# Format 3: Unix timestamp (integer or numeric string)
elif isinstance(raw, (int, float)):
dt = datetime.fromtimestamp(float(raw), tz=timezone.utc)
elif isinstance(raw, str) and raw.isdigit():
dt = datetime.fromtimestamp(int(raw), tz=timezone.utc)
if dt is None:
raise ValueError(
f"Cannot parse timestamp '{raw}' for field '{target_field}'"
)
record[target_field] = dt.astimezone(timezone.utc).isoformat()
return record
# Usage in any ingestion script — zero duplicate parsing logic:
users_pipeline = TransformPipeline()
users_pipeline.add(TransformStep("normalize_timestamp", normalize_timestamp))
orders_pipeline = TransformPipeline()
orders_pipeline.add(
TransformStep("normalize_order_date",
lambda r: normalize_timestamp(r, "order_date"))
)
# Each ingestion script imports and uses the shared pipeline:
# from transforms.pipeline import users_pipeline
# parsed_row = users_pipeline.run(raw_row)
When to apply: When ≥2 ingestion scripts or ETL jobs share the same transformation logic (timestamp parsing, currency conversion, locale formatting, ID normalization). Extract into a pipeline framework so new ingestors compose steps rather than reimplement them.
Pattern 3: API Response Canonicalization
Standardize response structures across all API endpoints using shared envelope types for pagination, errors, and metadata. Instead of each endpoint returning differently-shaped responses, every endpoint uses the same envelope schema enforced by base classes or generic types.
BAD — Each endpoint returns a different shape:
# ❌ BAD: users_view.py — Users endpoint with custom response shape
class UsersView(APIView):
def get(self, request):
users = User.objects.all()
page = int(request.GET.get("page", 1))
per_page = int(request.GET.get("per_page", 20))
start = (page - 1) * per_page
return Response({
"results": list(users[start:start + per_page].values()),
# Inconsistent key: uses "total" not "count"
"total": users.count(),
# Pagination uses different field names than other endpoints
"has_more": start + per_page < users.count(),
})
# ❌ BAD: orders_view.py — Orders endpoint with yet another shape
class OrdersView(APIView):
def get(self, request):
orders = Order.objects.all()
# Different pagination approach — uses offset/limit directly
offset = int(request.GET.get("offset", 0))
limit = int(request.GET.get("limit", 50))
return Response({
"data": list(orders[offset:offset + limit].values()),
"meta": {
"total_count": orders.count(), # nested under meta
"page": offset // limit + 1, # nested under meta
},
# No error shape defined here — errors just return {"detail": "..."}
})
# ❌ Bad: products_view.py — Products endpoint returns yet differently
class ProductsView(APIView):
def get(self, request):
products = Product.objects.all()
return Response({
"items": list(products.values()), # uses "items" key
"pagination": { # fully nested pagination object
"page": 1,
"perPage": 25, # camelCase! inconsistent
"totalPages": (products.count() + 24) // 25,
"hasNextPage": products.count() > 25,
},
})
# ❌ BAD: Error responses are also inconsistent across the API
# Users endpoint: {"errors": [{"field": "email", "message": "invalid"}]}
# Orders endpoint: {"detail": "Validation failed"}
# Products endpoint: {"error": "Not found"} — no field info at all
GOOD — Shared response envelope types used by every endpoint:
# ✅ GOOD: api/envelopes.py — Canonical response shapes for the entire API
from typing import Generic, TypeVar, ClassVar
from pydantic import BaseModel, Field
T = TypeVar("T")
class ApiError(BaseModel):
"""Canonical error response shape used by all endpoints."""
code: str = Field(description="Machine-readable error code")
message: str = Field(description="Human-readable error description")
field: str | None = Field(default=None, description="Field that caused the error")
details: dict | None = Field(default=None)
class ErrorResponse(BaseModel):
"""Standard error envelope — every 4xx/5xx returns this shape."""
errors: list[ApiError] = Field(
default_factory=list,
description="List of validation or runtime errors"
)
def add_error(
self, code: str, message: str, field: str | None = None, details: dict | None = None
) -> None:
"""Add a single error to the response."""
self.errors.append(ApiError(code=code, message=message, field=field, details=details))
class PaginatedResponse(BaseModel, Generic[T]):
"""Canonical paginated list envelope used by every list endpoint.
All endpoints return identical pagination structure so client libraries
can be written once and work across all resources.
"""
items: list[T] = Field(description="Page of results")
total_count: int = Field(description="Total number of matching records")
page: int = Field(description="Current page number (1-based)")
page_size: int = Field(description="Number of items per page")
has_next: bool = Field(description="Whether more pages exist after this one")
@classmethod
def from_queryset(
cls,
queryset,
item_serializer,
*,
page: int = 1,
page_size: int = 20,
) -> "PaginatedResponse":
"""Build a paginated response from a Django ORM queryset."""
total_count = queryset.count()
start = (page - 1) * page_size
items = queryset[start : start + page_size]
return cls(
items=[item_serializer(item).model_dump() for item in items],
total_count=total_count,
page=page,
page_size=page_size,
has_next=start + page_size < total_count,
)
class ApiResponse(BaseModel, Generic[T]):
"""Universal response envelope with optional data payload and metadata."""
success: bool = True
data: T | None = None
pagination: PaginatedResponse | None = None
errors: list[ApiError] | None = None
@classmethod
def for_list(
cls,
queryset,
item_serializer,
page: int = 1,
page_size: int = 20,
) -> "ApiResponse":
"""Convenience constructor for paginated list responses."""
return cls(
data=None,
pagination=PaginatedResponse.from_queryset(
queryset, item_serializer, page=page, page_size=page_size
),
)
# ✅ GOOD: users_view.py — Endpoint uses canonical envelope
class UsersView(APIView):
serializer = UserSerializer # assume this is defined elsewhere
def get(self, request):
users = User.objects.all()
page = int(request.GET.get("page", 1))
page_size = int(request.GET.get("per_page", 20))
response = ApiResponse.for_list(
queryset=users,
item_serializer=self.serializer,
page=page,
page_size=page_size,
)
return Response(response.model_dump(), status=status.HTTP_200_OK)
def post(self, request):
errors = ErrorResponse()
email = request.data.get("email", "")
if not re.match(r"^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$", email):
errors.add_error("INVALID_FORMAT", "Email must be a valid email address", field="email")
if errors.errors:
return Response(errors.model_dump(), status=status.HTTP_422_UNPROCESSABLE_ENTITY)
# ... create user logic ...
When to apply: When your API has ≥3 endpoints that shape their responses differently — different pagination keys, inconsistent error formats, or mixed success/error conventions. A shared envelope type eliminates the need for every client to handle multiple response shapes.
Pattern 4: Query Abstraction Layer
Extract complex repeated SQL query patterns into parameterized builder functions or CTE-based reusable queries. Instead of duplicating JOIN + WHERE chains across report generation functions, define a query assembly layer that composes shared fragments with service-specific filters.
BAD — Same JOIN pattern duplicated across four report functions:
-- ❌ BAD: reports/sales.py — Each function reimplements the same JOINs
SELECT o.id, o.total, u.name, u.email
FROM orders o
JOIN users u ON u.id = o.user_id
JOIN addresses a ON a.id = o.shipping_address_id
WHERE o.created_at >= '2024-01-01' AND o.created_at < '2025-01-01'
AND o.status != 'cancelled'
AND u.country_code = 'US'
ORDER BY o.total DESC;
-- ❌ BAD: reports/refunds.py — Same JOINs, slightly different WHERE clause
SELECT r.id, r.amount, u.name, u.email
FROM refunds r
JOIN orders o ON o.id = r.order_id
JOIN users u ON u.id = o.user_id
JOIN addresses a ON a.id = o.shipping_address_id -- same join, duplicated
WHERE r.created_at >= '2024-01-01' AND r.created_at < '2025-01-01'
AND o.status != 'cancelled' -- same filter, duplicated
AND u.country_code = 'US' -- same filter, duplicated
ORDER BY r.amount DESC;
-- ❌ BAD: reports/deliveries.py — Yet another copy with a new twist
SELECT d.id, d.tracking_number, o.total, u.name
FROM deliveries d
JOIN orders o ON o.id = d.order_id
JOIN users u ON u.id = o.user_id -- again, same join chain
JOIN addresses a ON a.id = o.shipping_address_id -- and again
WHERE d.status = 'delivered'
AND o.created_at >= '2024-06-01' -- different date range!
AND u.country_code IN ('US', 'CA') -- different country filter!
ORDER BY d.delivered_at;
-- ❌ BAD: reports/analytics.py — Fourth copy with yet more duplication
SELECT o.id, o.total, u.name, u.email, a.city, a.state_province
FROM orders o
JOIN users u ON u.id = o.user_id
JOIN addresses a ON a.id = o.shipping_address_id
WHERE o.status != 'cancelled'
AND o.created_at >= NOW() - INTERVAL '12 months'
AND u.country_code IN ('US', 'CA', 'GB')
GROUP BY o.id, u.name, u.email, a.city, a.state_province
HAVING COUNT(*) > 1;
GOOD — Parameterized query builder that assembles the common parts:
# ✅ GOOD: queries/report_builder.py — Canonical query assembly layer
from dataclasses import dataclass, field
from datetime import date
@dataclass
class ReportFilters:
"""Shared filter criteria used across all order-related reports."""
start_date: date | None = None
end_date: date | None = None
status_exclude: list[str] = field(default_factory=lambda: ["cancelled"])
countries: list[str] | None = None
def build_where_clauses(self) -> tuple[list[str], list[str]]:
"""Build parameterized WHERE clauses and their values.
Returns:
Tuple of (clause_strings, param_values) for safe query execution.
"""
clauses: list[str] = []
params: list[str] = []
if self.start_date is not None:
clauses.append("o.created_at >= %s")
params.append(self.start_date.isoformat())
if self.end_date is not None:
clauses.append("o.created_at < %s")
params.append(self.end_date.isoformat())
if self.status_exclude:
placeholders = ", ".join(f"%s" for _ in self.status_exclude)
clauses.append(f"o.status NOT IN ({placeholders})")
params.extend(self.status_exclude)
if self.countries:
placeholders = ", ".join(f"%s" for _ in self.countries)
clauses.append(f"u.country_code IN ({placeholders})")
params.extend(self.countries)
return clauses, params
def build_joins(self) -> str:
"""Return the canonical JOIN chain used by all order reports."""
return (
"JOIN users u ON u.id = o.user_id "
"JOIN addresses a ON a.id = o.shipping_address_id"
)
class ReportQueryBuilder:
"""Assembles parameterized report queries from shared fragments.
Each report specifies its SELECT columns and any report-specific
WHERE clauses, while the query builder handles all common JOINs
and filters.
"""
@staticmethod
def build(
select_columns: list[str],
base_table: str = "orders o",
extra_filters: list[str] | None = None,
extra_params: list[str] | None = None,
group_by: list[str] | None = None,
having_clause: str | None = None,
order_by: str | None = None,
limit: int | None = None,
filters: ReportFilters | None = None,
) -> tuple[str, list[str]]:
"""Build a complete parameterized report query.
Args:
select_columns: Columns to SELECT (e.g., ["o.id", "o.total"]).
base_table: Base table with alias (default: "orders o").
extra_filters: Report-specific WHERE clauses beyond shared filters.
extra_params: Values for report-specific filters.
group_by: GROUP BY columns if aggregating.
having_clause: HAVING clause for post-aggregation filter.
order_by: ORDER BY clause.
limit: Optional LIMIT value.
filters: Shared ReportFilters instance with date/status/country constraints.
Returns:
Tuple of (sql_string, param_values) ready for cursor.execute().
Raises:
ValueError: If select_columns is empty or filters are inconsistent.
"""
if not select_columns:
raise ValueError("select_columns must contain at least one column")
# Shared JOINs — defined once, used everywhere
join_clause = ""
where_clauses: list[str] = []
params: list[str] = []
joins_and_filters = ReportFilters() if filters is None else filters
# Build WHERE clauses from shared filters
shared_where, shared_params = joins_and_filters.build_where_clauses()
where_clauses.extend(shared_where)
params.extend(shared_params)
# Add report-specific filters on top
if extra_filters:
where_clauses.extend(extra_filters)
params.extend(extra_params or [])
where_sql = " AND ".join(where_clauses) if where_clauses else "TRUE"
query = (
f"SELECT {', '.join(select_columns)}\n"
f"FROM {base_table}\n"
f"{joins_and_filters.build_joins() if not joins_and_filters == ReportFilters() else ''}\n"
f"WHERE {where_sql}"
)
if group_by:
query += f"\nGROUP BY {', '.join(group_by)}"
if having_clause:
query += f"\nHAVING {having_clause}"
if order_by:
query += f"\nORDER BY {order_by}"
if limit is not None:
query += f"\nLIMIT %s"
params.append(str(limit))
return query, params
# ✅ GOOD: reports/sales.py — Uses the shared builder, zero JOIN duplication
def generate_sales_report(
start_date: date, end_date: date, countries: list[str] | None = None
) -> tuple[str, list[str]]:
"""Generate sales report using canonical query builder.
Only specifies what differs: SELECT columns and date range.
All JOINs, status exclusion, and country filtering are handled
by the shared QueryQueryBuilder layer.
"""
filters = ReportFilters(
start_date=start_date,
end_date=end_date,
countries=countries,
)
return ReportQueryBuilder.build(
select_columns=["o.id", "o.total", "u.name", "u.email"],
order_by="o.total DESC",
limit=1000,
filters=filters,
)
# ✅ GOOD: reports/refunds.py — Same builder, zero JOIN duplication
def generate_refund_report(
start_date: date, end_date: date, countries: list[str] | None = None
) -> tuple[str, list[str]]:
"""Generate refund report using canonical query builder."""
filters = ReportFilters(
start_date=start_date,
end_date=end_date,
countries=countries,
)
return ReportQueryBuilder.build(
select_columns=["r.id", "r.amount", "u.name", "u.email"],
base_table="refunds r", # different base table
extra_filters=["o.status != %s"], # extra filter on the joined orders table
extra_params=["cancelled"],
order_by="r.amount DESC",
filters=filters,
)
# ✅ GOOD: reports/analytics.py — Same builder with aggregation
def generate_order_analytics(
months_back: int = 12, countries: list[str] | None = None
) -> tuple[str, list[str]]:
"""Generate order analytics with grouping and having clause."""
filters = ReportFilters(
start_date=date.today().replace(month=max(1, date.today().month - months_back % 12 + 1)) if months_back <= 12 else None,
countries=countries,
)
return ReportQueryBuilder.build(
select_columns=["o.id", "o.total", "u.name", "a.city", "a.state_province"],
group_by=["o.id", "u.name", "a.city", "a.state_province"],
having_clause="COUNT(*) > 1",
filters=filters,
)
When to apply: When ≥2 query functions share the same JOIN chains or WHERE filter patterns but differ in SELECT columns, date ranges, or additional filters. The query builder handles all common fragments so report functions only specify their unique parts.
Pattern 5: Configuration Schema Normalization
Eliminate duplicated configuration validation by defining a base schema that every service extends. Instead of each service parsing and validating its own config from scratch with overlapping field definitions, use a shared base schema with per-service extension hooks for environment-specific overrides.
BAD — Each service validates config independently:
# ❌ BAD: services/analytics/config.py — Validates its own config
import os
from dataclasses import dataclass
@dataclass
class AnalyticsConfig:
"""Analytics service configuration — validates everything from scratch."""
@classmethod
def from_env(cls) -> "AnalyticsConfig":
db_host = os.environ.get("ANALYTICS_DB_HOST", "")
if not db_host or "@" in db_host:
raise ValueError("ANALYTICS_DB_HOST must be a valid hostname")
db_port_str = os.environ.get("ANALYTICS_DB_PORT", "5432")
try:
db_port = int(db_port_str)
if not (1 <= db_port <= 65535):
raise ValueError(f"Invalid port: {db_port}")
except ValueError as exc:
raise ValueError(f"ANALYTICS_DB_PORT must be integer 1-65535, got: {exc}")
db_name = os.environ.get("ANALYTICS_DB_NAME", "")
if not db_name:
raise ValueError("ANALYTICS_DB_NAME is required")
api_timeout_str = os.environ.get("ANALYTICS_API_TIMEOUT", "30")
try:
api_timeout = int(api_timeout_str)
if api_timeout < 1 or api_timeout > 300:
raise ValueError(f"Timeout must be 1-300s, got: {api_timeout}")
except ValueError as exc:
raise ValueError(f"ANALYTICS_API_TIMEOUT invalid: {exc}")
return cls(
db_host=db_host,
db_port=db_port,
db_name=db_name,
api_timeout=api_timeout,
)
# ❌ BAD: services/billing/config.py — Same validation pattern, different variable names
import os
from dataclasses import dataclass
@dataclass
class BillingConfig:
"""Billing service configuration — re-validates identical fields from scratch."""
@classmethod
def from_env(cls) -> "BillingConfig":
db_host = os.environ.get("BILLING_DB_HOST", "")
if not db_host or "@" in db_host:
raise ValueError("BILLING_DB_HOST must be a valid hostname") # same logic!
db_port_str = os.environ.get("BILLING_DB_PORT", "5432")
try:
db_port = int(db_port_str)
if not (1 <= db_port <= 65535):
raise ValueError(f"Invalid port: {db_port}") # same logic!
except ValueError as exc:
raise ValueError(f"BILLING_DB_PORT must be integer 1-65535: {exc}") # same logic!
db_name = os.environ.get("BILLING_DB_NAME", "")
if not db_name:
raise ValueError("BILLING_DB_NAME is required") # same logic!
api_timeout_str = os.environ.get("BILLING_API_TIMEOUT", "30")
try:
api_timeout = int(api_timeout_str)
if api_timeout < 1 or api_timeout > 300:
raise ValueError(f"Timeout must be 1-300s, got: {api_timeout}") # same logic!
except ValueError as exc:
raise ValueError(f"BILLING_API_TIMEOUT invalid: {exc}") # same logic!
# Plus billing-specific fields...
stripe_key = os.environ.get("STRIPE_SECRET_KEY", "")
if not stripe_key.startswith("sk_"):
raise ValueError("STRIPE_SECRET_KEY must start with sk_")
return cls(
db_host=db_host,
db_port=db_port,
db_name=db_name,
api_timeout=api_timeout,
stripe_key=stripe_key,
)
# ❌ BAD: services/inventory/config.py — Third copy of the same validation logic
import os
from dataclasses import dataclass
@dataclass
class InventoryConfig:
@classmethod
def from_env(cls) -> "InventoryConfig":
db_host = os.environ.get("INVENTORY_DB_HOST", "")
if not db_host or "@" in db_host: # SAME validation, AGAIN
raise ValueError("INVENTORY_DB_HOST must be a valid hostname")
db_port_str = os.environ.get("INVENTORY_DB_PORT", "5432")
try:
db_port = int(db_port_str)
if not (1 <= db_port <= 65535): # SAME validation, AGAIN
raise ValueError(f"Invalid port: {db_port}")
except ValueError as exc:
raise ValueError(f"INVENTORY_DB_PORT invalid: {exc}")
db_name = os.environ.get("INVENTORY_DB_NAME", "")
if not db_name: # SAME validation, AGAIN
raise ValueError("INVENTORY_DB_NAME is required")
api_timeout_str = os.environ.get("INVENTORY_API_TIMEOUT", "30")
try:
api_timeout = int(api_timeout_str)
if api_timeout < 1 or api_timeout > 300: # SAME validation, AGAIN
raise ValueError(f"Timeout must be 1-300s, got: {api_timeout}")
except ValueError as exc:
raise ValueError(f"INVENTORY_API_TIMEOUT invalid: {exc}")
return cls(...)
GOOD — Shared base config schema with per-service extensions:
# ✅ GOOD: config/schema.py — Canonical base configuration schema
import os
from dataclasses import dataclass, field, fields as dataclass_fields
from typing import Any, ClassVar
@dataclass
class BaseConfig:
"""Canonical shared configuration schema for all microservices.
Every service extends this class and adds only its own fields.
Validation logic lives in one place — the validate() method below.
Environment-specific overrides are supported via the environment_name field.
"""
# Shared fields present in every service
db_host: str
db_port: int
db_name: str
api_timeout: int
# Metadata for environment-aware configuration loading
environment_name: str = "development"
_VALID_PORT_RANGE: ClassVar[tuple[int, int]] = (1, 65535)
_TIMEOUT_RANGE: ClassVar[tuple[int, int]] = (1, 300)
@classmethod
def from_env(cls) -> "BaseConfig":
"""Load configuration from environment variables.
Extracts shared fields using the service-specific prefix pattern:
<SERVICE>_DB_HOST, <SERVICE>_DB_PORT, etc.
Returns:
Fully validated config instance for this service.
Raises:
ValueError: If any required field fails validation.
"""
service_prefix = cls._get_service_prefix()
db_host = os.environ.get(f"{service_prefix}_DB_HOST", "").strip()
db_port_str = os.environ.get(f"{service_prefix}_DB_PORT", "5432").strip()
db_name = os.environ.get(f"{service_prefix}_DB_NAME", "").strip()
api_timeout_str = os.environ.get(f"{service_prefix}_API_TIMEOUT", "30").strip()
# Parse shared values first
try:
db_port = cls._parse_int(db_port_str, f"{service_prefix}_DB_PORT")
except ValueError as exc:
raise ValueError(str(exc))
try:
api_timeout = cls._parse_int(api_timeout_str, f"{service_prefix}_API_TIMEOUT")
except ValueError as exc:
raise ValueError(str(exc))
config = cls(
db_host=db_host,
db_port=db_port,
db_name=db_name,
api_timeout=api_timeout,
…(truncated)