pgjinja 3.2.0

A Python package that seamlessly integrates PostgreSQL, Jinja templating, and Pydantic for type-safe database queries

pgjinja

pgjinja is a Python library for running PostgreSQL queries from Jinja2 SQL templates while keeping results type-safe with Pydantic models. It exists to separate SQL from application logic, reduce query string duplication, and provide a single sync/async API with connection pooling through psycopg3.

Tech Stack

• Language: Python 3.11+
• Build backend: Hatchling
• Core libraries: jinjasql2>=0.1.11, psycopg[binary,pool]>=3.2.6, pydantic>=2.10.6
• Dev tools: pytest, pytest-asyncio, pytest-cov, ruff, uv

How It Works

PgJinja and PgJinjaAsync are the two main clients. Both accept a PgJinjaSettings object that stores PostgreSQL credentials, template directory, and pool sizing. The sync client uses ConnectionPool and the async client uses AsyncConnectionPool.

When you call query(template, params, model), the library reads the SQL template file from template_dir, renders it with JinjaSql, and executes the prepared SQL with psycopg. For read queries, it returns tuples by default, or Pydantic model instances if you pass a model class.

If a model is provided, pgjinja injects _model_fields_ into template context so templates can select model-defined columns directly. This keeps SQL column lists aligned with your Pydantic schema without hand-maintaining SELECT fields in multiple places.

Both clients lazily open their connection pool on first use and retry failed query execution attempts (default: 2 attempts total), logging pool stats to help diagnose failures.

Getting Started

Prerequisites

• Python >=3.11
• [uv](https://docs.astral.sh/uv/)
• PostgreSQL (local or remote)

Installation

git clone https://github.com/tungph/pgjinja.git
cd pgjinja
uv sync

To install from PyPI instead:

pip install pgjinja

Configuration

The library is configured in Python through PgJinjaSettings:

Field	Required	Default	Description
user	Yes	-	PostgreSQL username
password	Yes	-	PostgreSQL password (SecretStr)
host	No	localhost	PostgreSQL host
port	No	5432	PostgreSQL port
dbname	No	postgres	Database name
template_dir	No	current directory	Directory containing SQL template files
min_size	No	4	Minimum pool size
max_size	No	None	Maximum pool size (None = unlimited)
application_name	No	pgjinja	Label for PostgreSQL connection logs

Example setup:

from pathlib import Path
from pydantic import SecretStr
from pgjinja import PgJinjaSettings, PgJinjaAsync

settings = PgJinjaSettings(
    user="postgres",
    password=SecretStr("postgres"),
    host="localhost",
    dbname="postgres",
    template_dir=Path("./templates"),
)

client = PgJinjaAsync(settings)

Application lifetime (singletons)

If you keep a single PgJinja or PgJinjaAsync instance for the whole process (typical for web apps and workers), its connection pool opens lazily on the first query. You should close the pool when the application shuts down so connections are released cleanly:

• Async: await client.close() in your teardown path (for example ASGI lifespan shutdown, or a finally block in asyncio.run()).
• Sync: client.close() on process exit or in your framework’s shutdown hook.

If the pool was opened and the client is destroyed without closing, you may see a log warning asking you to use await client.close() for async clients.

Example (FastAPI-style lifespan):

from contextlib import asynccontextmanager

from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    app.state.db = PgJinjaAsync(settings)
    yield
    await app.state.db.close()

app = FastAPI(lifespan=lifespan)

Running Locally

make run-example

examples/merchant_example.py expects a local examples/config.ini file with a [database] section (user, password, host, dbname) and uses SQL templates in examples/template/.

Verifying It Works

Run tests:

make test

Or run lint + formatting checks:

make lint

External Dependencies

Name	Purpose	Local Setup	Env Vars
PostgreSQL	Executes rendered SQL templates via psycopg3 pool connections	Run PostgreSQL locally (for example via Docker or system package) and create a database/user reachable by PgJinjaSettings	None required by library itself (credentials provided in PgJinjaSettings; example script reads examples/config.ini)

Project Structure

.
├── src/pgjinja/
│   ├── pgjinja.py                  # Sync client (ConnectionPool + template execution)
│   ├── pgjinja_async.py            # Async client (AsyncConnectionPool + template execution)
│   ├── schemas/pgjinja_settings.py # Typed DB and pool configuration model
│   └── shared/
│       ├── common.py               # Template reading and model-field helpers
│       └── execution.py            # Shared query prep and result mapping helpers
├── tests/                          # Unit/integration-style test suite
├── examples/
│   ├── merchant_example.py         # End-to-end async usage example
│   └── template/                   # Example SQL Jinja templates
├── pyproject.toml                  # Packaging, dependencies, pytest/ruff config
└── Makefile                        # Common development and release commands

Common Tasks

Task	Command
Install dependencies	make install
Run tests	make test
Lint + format	make lint
Format only	make format
Build package	make build
Run example app	make run-example
Publish to PyPI	make publish

Maintainer

@tungph