Database Connector Configuration for DSR Discovery Pipelines

Within the broader Cross-System Data Discovery & Sync architecture, the database connector is the read-only entry point that touches your most sensitive relational stores during a Data Subject Request (DSR). It is also the layer where configuration drift does the most damage: a connector that opens too many sessions, ignores a statement timeout, or accepts an unvalidated payload will silently corrupt the discovery manifest and blow through the response window GDPR Article 12(3) fixes at one month and CCPA §1798.130(a)(2) fixes at 45 days. This guide addresses one specific gap — how to configure a Postgres-family connector so that every extraction is deterministic, idempotent, credential-isolated, and provably bounded in time — before those records reach the Schema Validation Rules stage that gates fulfillment.

The connector is not a generic ORM session. It is a compliance control point that must fail closed, never mutate source data, and emit enough telemetry to reconstruct exactly what it read and when. The phases below map to the internal stages of that connector: secure initialization, pooled and time-budgeted execution, payload validation, async decoupling, and error routing with audit telemetry.

Phase 1: Secure Initialization and Credential Injection #

Connection strings must never reside in plaintext repositories or hardcoded configuration files — that requirement follows directly from the NIST SP 800-57 Part 1 key-management principle that secret material be isolated from the code that consumes it. Standardize credential injection through an environment-bound secrets manager, and lean on Python’s type system to prevent runtime coercion failures that would otherwise surface mid-extraction.

Typed configuration objects eliminate ambiguity during pool instantiation. Using dataclasses, we enforce strict attribute typing and default fallbacks for non-sensitive parameters. Note that os.getenv() is evaluated at class-definition time here, so the process environment must be populated before the module is imported — inject secrets in the entrypoint, not lazily.

import os
from dataclasses import dataclass

@dataclass(frozen=True)
class DSRConnectorConfig:
    host: str = os.getenv("DB_HOST", "")
    port: int = int(os.getenv("DB_PORT", "5432"))
    database: str = os.getenv("DB_NAME", "")
    user: str = os.getenv("DB_USER", "")
    password: str = os.getenv("DB_PASS", "")
    max_connections: int = 10
    connection_timeout: int = 5
    statement_timeout_ms: int = 30000
    ssl_mode: str = "require"
    application_name: str = "dsr-pipeline-worker"

The frozen=True parameter guarantees immutability post-instantiation, preventing accidental credential mutation during pipeline execution. Setting application_name also tags every session in pg_stat_activity, so an auditor can attribute discovery queries to the DSR worker rather than to an anonymous connection.

The table below documents the compliance-relevant parameters and why each default is chosen:

Phase 2: Connection Pooling and Execution Budgets #

Connection pooling must align with query execution budgets to prevent resource exhaustion during bulk extraction. Enforce strict cursor isolation levels and apply the server-side statement_timeout from the config so that no single query can outlive its budget.

Using psycopg2, configure a ThreadedConnectionPool with explicit timeout boundaries. The context manager below sets the statement timeout on each checkout and guarantees cursor closure and transaction rollback on any exception:

import psycopg2
from contextlib import contextmanager
from psycopg2.pool import ThreadedConnectionPool
from psycopg2.extensions import ISOLATION_LEVEL_REPEATABLE_READ

def init_pool(cfg: DSRConnectorConfig) -> ThreadedConnectionPool:
    return ThreadedConnectionPool(
        minconn=2,
        maxconn=cfg.max_connections,
        host=cfg.host,
        port=cfg.port,
        dbname=cfg.database,
        user=cfg.user,
        password=cfg.password,
        connect_timeout=cfg.connection_timeout,
        sslmode=cfg.ssl_mode,
        application_name=cfg.application_name,
        options=f"-c statement_timeout={cfg.statement_timeout_ms}",
    )

@contextmanager
def acquire_cursor(pool: ThreadedConnectionPool):
    conn = pool.getconn()
    conn.set_isolation_level(ISOLATION_LEVEL_REPEATABLE_READ)
    conn.set_session(readonly=True)
    cursor = conn.cursor()
    try:
        yield cursor
        conn.commit()
    except Exception:
        conn.rollback()
        raise
    finally:
        cursor.close()
        pool.putconn(conn)

Repeatable-read isolation prevents phantom reads during long-running DSR extractions, so the manifest reflects a single consistent snapshot rather than a moving target. Calling set_session(readonly=True) enforces the read-only contract at the transport layer — the connector physically cannot mutate source records, which is exactly the guarantee an auditor wants to see. When scaling across heterogeneous data stores, teams often adapt these pool abstractions using Connecting PostgreSQL and Snowflake for DSR discovery to reach cloud-native warehouses without breaking the timeout contract.

Phase 3: Schema Validation and Circuit Breaking #

Schema validation gates every extraction phase. Deploy Pydantic v2 models to enforce column presence, data-type conformity, and payload completeness; invalid payloads trip an immediate circuit breaker so malformed records never reach the transformation layer. This connector-side check is the first half of a contract that the dedicated Schema Validation Rules stage completes downstream.

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator
from typing import Any
from datetime import datetime

class DSRExtractionSchema(BaseModel):
    model_config = ConfigDict(extra="forbid", frozen=True)

    subject_id: str = Field(..., min_length=1)
    pii_fields: list[str] = Field(..., min_length=1)
    extraction_timestamp: datetime
    row_count: int = Field(..., ge=0)
    metadata: dict[str, Any] = Field(default_factory=dict)
    compliance_tag: str | None = None

    @field_validator("pii_fields")
    @classmethod
    def normalize_pii_fields(cls, v: list[str]) -> list[str]:
        return [f.strip().lower() for f in v if f.strip()]

def validate_extraction(payload: dict[str, Any]) -> DSRExtractionSchema:
    try:
        return DSRExtractionSchema(**payload)
    except ValidationError as e:
        raise RuntimeError(f"Schema violation: {e}") from e

Leveraging Pydantic v2 ensures extraction outputs conform to compliance-defined contracts before they enter the transformation layer. The extra="forbid" setting rejects any unexpected column so schema drift in the source surfaces as a loud failure rather than a silent data leak, and the normalize_pii_fields validator guarantees consistent casing and whitespace — critical for the downstream deduplication and redaction routines that the PII Extraction & Redaction Pipelines stage performs.

Phase 4: Cross-Environment Type Coercion and Async Decoupling #

Cross-environment mapping requires explicit type-coercion matrices. Normalize temporal definitions across hybrid environments with deterministic casting rules that handle timezone drift, epoch conversions, and legacy VARCHAR date formats — an inconsistent extraction_timestamp will misalign the SLA clock and undermine the audit trail.

Asynchronous execution decouples heavy extraction jobs from the main pipeline thread. Pair this connector with Async Polling & Queue Management to maintain predictable throughput and prevent blocking I/O during large-scale subject lookups. The pattern below offloads blocking psycopg2 calls to a thread-pool executor so the event loop stays free, and it uses asyncio.get_running_loop() rather than the deprecated get_event_loop():

import asyncio
from typing import AsyncGenerator

async def async_extraction_worker(
    pool: ThreadedConnectionPool, query: str, params: tuple
) -> AsyncGenerator[dict, None]:
    loop = asyncio.get_running_loop()
    with acquire_cursor(pool) as cursor:
        await loop.run_in_executor(None, cursor.execute, query, params)
        col_names = [desc[0] for desc in cursor.description]
        while True:
            rows = await loop.run_in_executor(None, cursor.fetchmany, 500)
            if not rows:
                break
            for row in rows:
                yield dict(zip(col_names, row))

The fetchmany batch size aligns with memory budgets, preventing out-of-memory conditions during high-volume DSR sweeps, and async execution keeps the connector responsive to health checks and cancellation signals should a request be withdrawn mid-flight.

Phase 5: Error Routing and Compliance Telemetry #

Validation failures route to a dedicated error-categorization queue so that malformed records never contaminate compliance reports. Track rejection rates against SLA thresholds and maintain append-only audit logs for regulatory review under GDPR Art. 5(2) accountability.

Retry logic must be idempotent and bounded. Apply exponential backoff with jitter to transient network failures, while permanent schema violations trigger immediate dead-letter routing. Unified retry patterns across database and API connectors are documented in SaaS API Sync Strategies, ensuring consistent failure handling regardless of the underlying transport.

Compliance hooks are embedded directly into the connector lifecycle:

• Extraction start/end timestamps — logged for audit-trail reconstruction and SLA attribution.
• Row-count verification — cross-checked against source metadata to detect truncation.
• PII field mapping — validated against the organization’s data-classification matrix before export.
• Circuit-breaker state — exported as a Prometheus metric that halts the pipeline when the error rate exceeds 2%.

Edge Cases and Conflict Resolution #

Real estates rarely behave like the happy path. The connector must resolve these deterministically rather than guessing:

• Pool exhaustion under concurrent sweeps. When getconn() blocks because all sessions are checked out, fail fast with a bounded wait and route the task back to the queue rather than holding the event loop hostage. Never raise maxconn above what the source can tolerate; degrading production OLTP to satisfy a DSR is itself a compliance risk.
• Statement timeout versus large legitimate scans. A wide subject match on a high-cardinality table can legitimately exceed 30 s. Resolve by keyset-paginating the query rather than lifting the timeout — the timeout is a control, not a nuisance.
• Conflicting timestamps across replicas. Reading from a lagging read-replica can yield an extraction_timestamp earlier than the request. Pin discovery to a snapshot LSN, or reject rows whose timestamp precedes the attested request time.
• UNKNOWN classification tags. If a column is not present in the data-classification matrix, treat it as UNKNOWN and fail closed to the dead-letter queue for manual triage — never default an unlabeled field to non-PII.
• Duplicate identifiers across shards. The same subject_id surfacing on multiple shards is expected; hash and deduplicate at the manifest layer so the count reflects distinct records, not distinct connections.

Performance and Scale Considerations #

Discovery latency is a compliance metric, not just an SRE one — a slow connector eats the statutory window. Tune for predictable throughput:

• Pool sizing. Set maxconn from the source’s headroom, not the worker’s ambition. Two-to-ten sessions per worker is typical; measure pg_stat_activity under load before raising it.
• Batch cursor fetches. A fetchmany(500) batch keeps memory flat and lets backpressure propagate; server-side cursors (cursor("named")) avoid materializing the full result set for very large matches.
• Cache resolved identifiers. Memoize subject-to-shard resolution in Redis for the life of a request so repeated lookups within one DSR do not re-hit every store.
• Isolate consumers. Give discovery its own database role and connection budget, separate from ingestion, so a burst of requests cannot starve unrelated workloads.
• Throughput target. Aim for a discovery pass that completes in minutes, not hours; anything approaching the 30-day GDPR boundary should already have paged an operator.

Testing and Compliance Verification #

Treat the connector like any other control: prove it behaves before it touches production PII.

• Payload matrix. Feed validate_extraction a matrix of valid, missing-field, extra-field, wrong-type, and empty-pii_fields payloads; assert the last four raise and only the first returns a model.
• Read-only assertion. In an integration test, attempt a write through acquire_cursor and assert it raises — proving set_session(readonly=True) is enforced.
• Timeout regression. Run a deliberately slow query (SELECT pg_sleep(60)) and assert it aborts at statement_timeout_ms, guarding against a config regression silently disabling the cap.
• Held-out jurisdictions. Include test fixtures for regions your production data does not yet cover (for example an EU subject when your traffic is US-only) so jurisdiction handling is exercised before it is needed live.
• Telemetry assertion. Verify that a rejected payload increments the error-rate metric and writes an audit-log entry, so the circuit breaker and the accountability trail are both observable.

import pytest

def test_forbids_unknown_columns():
    payload = {
        "subject_id": "S1", "pii_fields": ["Email"],
        "extraction_timestamp": "2026-07-01T00:00:00Z",
        "row_count": 1, "leaked_column": "oops",
    }
    with pytest.raises(RuntimeError):
        validate_extraction(payload)

Frequently Asked Questions #

Why enforce a server-side statement timeout instead of a client-side one? #

A client-side timeout only stops the client from waiting — the query keeps running on the server, holding locks and consuming resources against the source. Setting statement_timeout via the connection options makes PostgreSQL itself abort the query, which is the only way to guarantee a query cannot outlive its budget during a high-volume sweep.

Does read-only isolation weaken data integrity for discovery? #

No — it strengthens it. A DSR discovery pass must never mutate source records, and set_session(readonly=True) combined with repeatable-read isolation gives you a consistent snapshot that the connector is physically incapable of changing. That combination is both a correctness guarantee and an audit-friendly control under GDPR Art. 5(1)(f).

How does this connector coordinate with schema validation downstream? #

The Pydantic gate here is a fast, connector-local rejection of obviously malformed rows. The authoritative contract lives in the Schema Validation Rules stage, which validates the full manifest against the DSR ingestion schema before fulfillment. Keeping both means bad data is caught early and again at the boundary.

What happens when the connection pool is exhausted mid-request? #

The task should fail fast with a bounded wait and be re-queued rather than blocking. Raising maxconn to paper over exhaustion risks degrading production OLTP, which is itself a compliance and availability problem. Size the pool from the source’s measured headroom and let Async Polling & Queue Management absorb the backpressure.

Which regulations shape these configuration defaults? #

Encryption in transit (ssl_mode=require) maps to GDPR Art. 32(1)(a); the bounded response window driving the timeout and throughput targets comes from GDPR Art. 12(3) and CCPA §1798.130(a)(2); credential isolation follows NIST SP 800-57 key-management guidance; and the append-only audit logging supports GDPR Art. 5(2) accountability.

Related #

• Schema Validation Rules — the ingestion contract that validates the connector’s output before fulfillment.
• SaaS API Sync Strategies — unified retry, backoff, and credential patterns for non-database sources.
• Async Polling & Queue Management — bounded concurrency and backpressure for large discovery sweeps.
• Connecting PostgreSQL and Snowflake for DSR discovery — adapting these pool abstractions to a cloud-native warehouse.
• Up to Cross-System Data Discovery & Sync — the discovery and synchronization architecture this connector feeds.