Forem: krylosov-aa

pg-status - a lightweight microservice for checking PostgreSQL host status

krylosov-aa — Sun, 04 Jan 2026 22:54:32 +0000

Hi! I’d like to introduce my new project — pg-status.

It’s a lightweight, high-performance microservice designed to determine the status of PostgreSQL hosts. Its main goal is to help your backend identify a live master and a sufficiently up-to-date synchronous replica.

Key features

Very easy to deploy as a sidecar and integrate with your existing PostgreSQL setup
Identifies the master and synchronous replicas, and assists with failover
Helps balance load between hosts

If you find this project useful, I’d really appreciate your support — a star on GitHub would mean a lot!

But first, let’s talk about the problem pg-status is built to solve.

PostgreSQL on multiple hosts

To improve the resilience and scalability of a PostgreSQL database, it’s common to run multiple hosts using the classic master–replica setup. There’s one master host that accepts writes, and one or more replicas that receive changes from the master via physical or logical replication.

Everything works great in theory — but there are a few important details to consider:

Any host can fail
A replica may need to take over as the master (failover)
A replica can significantly lag behind the master

From the perspective of a backend application connecting to these databases, this introduces several practical challenges:

How to determine which host is currently the live master
How to identify which replicas are available
How to measure replica lag to decide whether it’s suitable for reads
How to switch the client connection pool (or otherwise handle reconnection) after failover
How to distribute load effectively among hosts

There are already various approaches to solving these problems — each with its own pros and cons. Here are a few of the common methods I’ve encountered:

Via DNS

In this approach, specific hostnames point to the master and replica instances. Essentially, there’s no built-in master failover handling, and it doesn’t help determine the replica status — you have to query it manually via SQL.

It’s possible to add an external service that detects host states and updates the DNS records accordingly, but there are a few drawbacks:

DNS updates can take several seconds — or even tens of seconds — which can be critical
DNS might automatically switch to read-only mode

Overall, this solution does work, and pg-status can actually serve as such a service for host state detection.

Also, as far as I know, many PostgreSQL cloud providers rely on this exact mechanism.

Multihost in libpq

With this method, the client driver (libpq) can locate the first (or random) available host from a given list that matches the desired role (master or replica).

A change in the master is detected only after an actual SQL query fails — at which point the connection crashes, and the client cycles through the hosts list again upon reconnection.

Proxy

You can set up a proxy that supports on-the-fly configuration updates. In that case, you’ll also need some component responsible for notifying the proxy when it should switch to a different host.

This is generally a solid approach, but it still depends on an external mechanism that monitors PostgreSQL host states and communicates those changes to the proxy. pg-status fits perfectly for this purpose — it can serve as that mechanism.

Alternatively, you can use pgpool-II, which is specifically designed for such scenarios. It not only determines which host to route traffic to but can even perform automatic failover itself. The main downside, however, is that it can be complex to deploy and configure.

CloudNativePG

As far as I know, CloudNativePG already provides all this functionality out of the box. The main considerations here are deployment complexity and the requirement to run within a Kubernetes environment.

My solution - pg-status

At my workplace, we use a PostgreSQL cloud provider that offers a built-in failover mechanism and lets us connect to the master via DNS. However, I wanted to avoid situations where DNS updates take too long to reflect the new master.

I also wanted more control — not just connecting to the master, but also balancing read load across replicas and understanding how far each replica lags behind the master. At the same time, I didn’t want to complicate the system architecture with a shared proxy that could become a single point of failure.

In the end, the ideal solution turned out to be a tiny sidecar service running next to the backend. This sidecar takes responsibility for selecting the appropriate host. On the backend side, I maintain a client connection pool and, before issuing a connection, I check the current host status and immediately reconnect to the right one if needed.

The sidecar approach brings some extra benefits:

A sidecar failure affects only the single instance it’s attached to, not the entire system.
PostgreSQL availability is measured relative to the local instance — meaning the health check can automatically report that this instance shouldn't receive traffic if the database is unreachable (for example, due to network isolation between data centers).

That’s how pg-status was born. Its job is to periodically poll PostgreSQL hosts, keep track of their current state, and expose several lightweight, fast endpoints for querying this information.

You can call pg-status directly from your backend on each request — for example, to make sure the master hasn’t failed over, and if it has, to reconnect automatically. Alternatively, you can use its special endpoints to select an appropriate replica for read operations based on replication lag.

For example, I have a library for Python - context-async-sqlalchemy, which has a special place, where you can user pg-status to always get to the right host.

How to use

Installation

You can build pg-status from source, install it from a .deb or binary package, or run it as a Docker container (lightweight Alpine-based images are available or ubuntu-based). Currently, the target architecture is Linux amd64, but the microservice can be compiled for other targets using CMake if needed.

Usage

The service’s behavior is configured via environment variables. Some variables are required (for example, connection parameters for your PostgreSQL hosts), while others are optional and have default values.

You can find the full list of parameters here: https://github.com/krylosov-aa/pg-status?tab=readme-ov-file#parameters

When running, pg-status exposes several simple HTTP endpoints:

GET /master - returns the current master
GET /replica - returns a random replica using the round-robin algorithm
GET /sync_by_time - returns a synchronous replica based on time or the master, meaning the lag behind the master is measured in time
GET /sync_by_bytes - returns a synchronous replica based on bytes (based on the WAL LSN log) or the master, meaning the lag behind the master is measured in bytes written to the log
GET /sync_by_time_or_bytes - essentially a host from sync_by_time or from sync_by_bytes
GET /sync_by_time_and_bytes - essentially a host from sync_by_time and From sync_by_bytes
GET /hosts - returns a list of all hosts and their current status: live, master, or replica.

As you can see, pg-status provides a flexible API for identifying the appropriate replica to use. You can also set maximum acceptable lag thresholds (in time or bytes) via environment variables.

Almost all endpoints support two response modes:

Plain text (default)
JSON — when you include the header Accept: application/json For example: {"host": "localhost"}

pg-status can also work alongside a proxy or any other solution responsible for handling database connections. In this setup, your backend always connects to a single proxy host (for instance, one that points to the master). The proxy itself doesn’t know the current PostgreSQL state — instead, it queries pg-status via its HTTP endpoints to decide when to switch to a different host.

pg-status Implementation Details

pg-status is a microservice written in C. I chose this language for two main reasons:

It’s extremely resource-efficient — perfect for a lightweight sidecar scenario
I simply enjoy writing in C, and this project felt like a natural fit

The microservice consists of two core components running in two active threads:

1. PG Monitoring

The first thread is responsible for monitoring. It periodically polls all configured hosts using the libpq library to determine their current status. This part has an extensive list of configurable parameters, all set via environment variables:

How often to poll hosts
Connection timeout for each host
Number of failed connection attempts before marking a host as dead
Maximum acceptable replica lag (in milliseconds) considered “synchronous”
Maximum acceptable replica lag (in bytes, based on WAL LSN) considered “synchronous”

Currently, only physical replication is supported.

2. HTTP Server

The second thread runs the HTTP server, which handles client requests and retrieves the current host status from memory. It’s implemented using libmicrohttpd, offering great performance while keeping the footprint small.

This means your backend can safely query pg-status before every SQL operation without noticeable overhead.

In my testing (in a Docker container limited to 0.1 CPU and 6 MB of RAM), I achieved around 1500 RPS with extremely low latency. You can see detailed performance metrics here.

Potential Improvements

Right now, I’m happy with the functionality — pg-status is already used in production in my own projects. That said, some improvements I’m considering include:

Support for logical replication
Adding precise time and byte lag information directly to the JSON responses so clients can make more informed decisions

If you find the project interesting or have ideas for enhancements, feel free to open an issue on GitHub — contributions and feedback are always welcome!

Summary

pg-status is a lightweight, efficient microservice designed to solve a practical problem — determining the status of PostgreSQL hosts — while being exceptionally easy to deploy and operate.

Licensed under MIT
Open source and available on GitHub: https://github.com/krylosov-aa/pg-status
Available as source, .deb binary package, or Docker container

If you like the project, I’d really appreciate your support — please ⭐ it on GitHub!

Thanks for reading!

context-async-sqlalchemy - The best way to use sqlalchemy in an async python application

krylosov-aa — Tue, 09 Dec 2025 09:49:17 +0000

Hello! I’d like to introduce my new library - context-async-sqlalchemy. It makes working with SQLAlchemy in asynchronous Python applications incredibly easy. The library requires minimal code for simple use cases, yet offers maximum flexibility for more complex scenarios.

Let’s briefly review the theory behind SQLAlchemy - what it consists of and how it integrates into a Python application. We’ll explore some of the nuances and see how context-async-sqlalchemy helps you work with it more conveniently. Note that everything here refers to asynchronous Python.

Short Summary of SQLAlchemy

SQLAlchemy provides an Engine, which manages the database connection pool, and a Session, through which SQL queries are executed. Each session uses a single connection that it obtains from the engine.

The engine should have a long lifespan to keep the connection pool active. Sessions, on the other hand, should be short-lived, returning their connections to the pool as quickly as possible.

(In the diagram, both engines are shown connected to the same database - which is admittedly unusual. The goal is simply to illustrate that each engine has its own connection pool. In practice, your application will typically use one engine per database - for example, one for the master and another for the replica.)

Integration and Usage in an Application

Direct Usage

Let’s start with the simplest manual approach - using only SQLAlchemy, which can be integrated anywhere.

Create an engine and a session maker:

engine = create_async_engine(DATABASE_URL)

session_maker = async_sessionmaker(engine, expire_on_commit=False)

Now imagine we have an endpoint for creating a user:

@app.post("/users/")
async def create_user(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

On line 2, we open a session; on line 3, we begin a transaction; and finally, on line 4, we execute some SQL to create a user.

Now imagine that, as part of the user creation process, we need to execute two SQL queries:

@app.post("/users/")
async def create_user(name):
    await insert_user(name)
    await insert_user_profile(name)

async def insert_user(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

async def insert_user_profile(name):
    async with session_maker() as session:
        async with session.begin():
            await session.execute(stmt)

Here we encounter two problems:

Two transactions are being used, even though we probably want only one.
Code duplication.

We can try to fix this by moving the context managers to a higher level:

@app.post("/users/")
async def create_user(name:):
    async with session_maker() as session:
        async with session.begin():
            await insert_user(name, session)
            await insert_user_profile(name, session)

async def insert_user(name, session):
    await session.execute(stmt)

async def insert_user_profile(name, session):
    await session.execute(stmt)

But if we look at multiple handlers, the duplication still remains:

@app.post("/dogs/")
async def create_dog(name):
    async with session_maker() as session:
        async with session.begin():
            ...

@app.post("/cats")
async def create_cat(name):
    async with session_maker() as session:
        async with session.begin():
            ...

Dependency Injection

You can move session and transaction management into a dependency. For example, in FastAPI:

async def get_atomic_session():
    async with session_maker() as session:
        async with session.begin():
            yield session


@app.post("/dogs/")
async def create_dog(name, session = Depends(get_atomic_session)):
    await session.execute(stmt)


@app.post("/cats/")
async def create_cat(name, session = Depends(get_atomic_session)):
    await session.execute(stmt)

Code duplication is gone, but now the session and transaction remain open until the end of the request lifecycle, with no way to close them early and release the connection back to the pool.

This could be solved by returning a DI container from the dependency that manages sessions - however, that approach adds complexity, and no ready‑made solutions exist.

Additionally, the session now has to be passed through multiple layers of function calls, even to those that don’t directly need it:

@app.post("/some_handler/")
async def some_handler(session = Depends(get_atomic_session)):
    await do_first(session)
    await do_second(session)

async def do_first(session):
    await do_something()
    await insert_to_database(session)

async def insert_to_database(session):
    await session.execute(stmt)

As you can see, do_first doesn’t directly use the session but still has to accept and pass it along. Personally, I find this inelegant - I prefer to encapsulate that logic inside insert_to_database. It’s a matter of taste and philosophy.

Wrappers Around SQLAlchemy

There are various wrappers around SQLAlchemy that offer convenience but introduce new syntax - something I find undesirable. Developers already familiar with SQLAlchemy shouldn’t have to learn an entirely new API.

The New Library

I wasn’t satisfied with the existing approaches. In my FastAPI service, I didn’t want to write excessive boilerplate just to work comfortably with SQL. I needed a minimal‑code solution that still allowed flexible session and transaction control - but couldn’t find one. So I built it for myself, and now I’m sharing it with the world.

My goals for the library were:

Minimal boilerplate and no code duplication
Automatic commit or rollback when manual control isn’t required
The ability to manually manage sessions and transactions when needed
Suitable for both simple CRUD operations and complex logic
No new syntax - pure SQLAlchemy
Framework‑agnostic design

Here’s the result.

Simplest Scenario

To make a single SQL query inside a handler - without worrying about sessions or transactions:

from context_async_sqlalchemy import db_session

async def some_func() -> None:
    session = await db_session(connection)  # new session
    await session.execute(stmt)  # some sql query

    # commit automatically

The db_session function automatically creates (or reuses) a session and closes it when the request ends.

Multiple queries within one transaction:

@app.post("/users/")
async def create_user(name):
    await insert_user(name)
    await insert_user_profile(name)

async def insert_user(name):
    session = await db_session(connection)  # creates a session
    await session.execute(stmt)  # opens a connection and a transaction

async def insert_user_profile(name):
    session = await db_session(connection)  # gets the same session
    await session.execute(stmt)  # uses the same connection and transaction

Early Commit

Need to commit early? You can:

async def manual_commit_example():
    session = await db_session(connect)
    await session.execute(stmt)
    await session.commit()  # manually commit the transaction

Or, for example, consider the following scenario: you have a function called insert_something that’s used in one handler where an autocommit at the end of the query is fine. Now you want to reuse insert_something in another handler that requires an early commit. You don’t need to modify insert_something at all - you can simply do this:

async def example_1():
    await insert_something()  # autocommit is suitable for us here

async def example_2():
    await insert_something()  # here we want to make a commit before the update
    await commit_db_session(connect)  # commits the context transaction
    await update_something()  # works with a new transaction

Or, even better, you can do it this way - by wrapping the function in a separate transaction:

async def example_2():
    async with atomic_db_session(connect):
        # a transaction is opened and closed
        await insert_something()

    await update_something()  # works with a new transaction

You can also perform an early rollback using rollback_db_session.

Early Session Close

There are situations where you may need to close a session to release its connection - for example, while performing other long‑running operations.
You can do it like this:

async def example_with_long_work():
    async with atomic_db_session(connect):
        await insert_something()

    await close_db_session(connect)  # released the connection

    ...
    # some very long work here
    ...

    await update_something()

close_db_session closes the current session. When update_something calls db_session, it will already have a new session with a different connection.

Concurrent Queries

In SQLAlchemy, you can’t run two concurrent queries within the same session. To do so, you need to create a separate session.

async def concurent_example():
    asyncio.gather(
        insert_something(some_args),
        insert_another_thing(some_args),  # error!
    )

The library provides two simple ways to execute concurrent queries.

async def concurent_example():
    asyncio.gather(
        insert_something(some_args),
        run_in_new_ctx(  # separate session with autocommit
            insert_another_thing, some_args
        ),
    )

run_in_new_ctx runs a function in a new context, giving it a fresh session. This can be used, for example, with functions executed via asyncio.gather or asyncio.create_task.

Alternatively, you can work with a session entirely outside of any context - just like in the manual mode described at the beginning.

async def insert_another_thing(some_args):
    async with new_non_ctx_session(connection) as session:
        await session.execute(stmt)
        await session.commit()

# or

async def insert_something(some_args):
    async with new_non_ctx_atomic_session(connection) as session:
        await session.execute(stmt)

These methods can be combined:

await asyncio.gather(
    _insert(),  # context session
    run_in_new_ctx(_insert),  # new context session
    _insert_non_ctx(),  # own manual session
)

Other Scenarios

The repository includes several application integration examples. You can also explore various scenarios for using the library. These scenarios also serve as tests for the library - verifying its behavior within a real application context rather than in isolation.

Integrating the Library with Your Application

Now let’s look at how to integrate this library into your application. The goal was to make the process as simple as possible.

We’ll start by creating the engine and session_maker, and by addressing the connect parameter, which is passed throughout the library functions. The DBConnect class is responsible for managing the database connection configuration.

from context_async_sqlalchemy import DBConnect

connection = DBConnect(
    engine_creator=create_engine,
    session_maker_creator=create_session_maker,
    host="127.0.0.1",
)

The intended use is to have a global instance responsible for managing the lifecycle of the engine and session_maker.

It takes two factory functions as input:

engine_creator - a factory function for creating the engine
session_maker_creator - a factory function for creating the session_maker

Here are some examples:

def create_engine(host):
    pg_user = "krylosov-aa"
    pg_password = ""
    pg_port = 6432
    pg_db = "test"
    return create_async_engine(
        f"postgresql+asyncpg://"
        f"{pg_user}:{pg_password}"
        f"@{host}:{pg_port}"
        f"/{pg_db}",
        future=True,
        pool_pre_ping=True,
    )

def create_session_maker(engine):
    return async_sessionmaker(
        engine, class_=AsyncSession, expire_on_commit=False
    )

host is an optional parameter that specifies the database host to connect to.

Why is the host optional, and why use factories? Because the library allows you to reconnect to the database at runtime - which is especially useful when working with a master and replica setup.

DBConnect also has another optional parameter - a handler that is called before creating a new session. You can place any custom logic there, for example:

async def renew_master_connect(connect: DBConnect):
    master_host = await get_master() # determine the master host

    if master_host != connect.host:  # if the host has changed
        await connect.change_host(master_host)  # reconnecting


master = DBConnect(
    ...

    # handler before session creation
    before_create_session_handler=renew_master_connect,
)

replica = DBConnect(
    ...
    before_create_session_handler=renew_replica_connect,
)

At the end of your application's lifecycle, you should gracefully close the connection. DBConnect provides a close() method for this purpose.

@asynccontextmanager
async def lifespan(app):
    # some application startup logic

    yield

    # application termination logic
    await connection.close()  # closing the connection to the database

All the important logic and “magic” of session and transaction management is handled by the middleware - and it’s very easy to set up.

Here’s an example for FastAPI:

from context_async_sqlalchemy.fastapi_utils import (
    add_fastapi_http_db_session_middleware,
)

app = FastAPI(...)
add_fastapi_http_db_session_middleware(app)

There is also pure ASGI middleware.

from context_async_sqlalchemy import ASGIHTTPDBSessionMiddleware

app.add_middleware(ASGIHTTPDBSessionMiddleware)

Testing

Testing is a crucial part of development. I prefer to test using a real, live PostgreSQL database. In this case, there’s one key issue that needs to be addressed - data isolation between tests. There are essentially two approaches:

Clearing data between tests. In this setup, the application uses its own transaction, and the test uses a separate one.
Using a shared transaction between the test and the application and performing rollbacks to restore the state.

The first approach is very convenient for debugging, and sometimes it’s the only practical option - for example, when testing complex scenarios involving multiple transactions or concurrent queries. It’s also a “fair” testing method because it checks how the application actually handles sessions.

However, it has a downside: such tests take longer to run because of the time required to clear data between them - even when using TRUNCATE statements, which still have to process all tables.

The second approach, on the other hand, is much faster thanks to rollbacks, but it’s not as realistic since we must prepare the session and transaction for the application in advance.

Don't forget to give the project a star on GitHub if you like it! Thanks!

In my projects, I use both approaches together: a shared transaction for most tests with simple logic, and separate transactions for the minority of more complex scenarios.

The library provides a few utilities that make testing easier. The first is rollback_session - a session that is always rolled back at the end. It’s useful for both types of tests and helps maintain a clean, isolated test environment.

@pytest_asyncio.fixture
async def db_session_test():
    async with rollback_session(master) as session:
        yield session

For tests that use shared transactions, the library provides two utilities: set_test_context and put_savepoint_session_in_ctx.

@pytest_asyncio.fixture(autouse=True)
async def db_session_override(db_session_test):
    async with set_test_context():
        async with put_savepoint_session_in_ctx(master, db_session_test):
            yield

This fixture creates a context in advance, so the application runs within it instead of creating its own. The context also contains a pre‑initialized session that creates a release savepoint instead of performing a commit.

How it all works

Here's a diagram of how it all works:

The middleware initializes the context, and your application accesses it through the library’s functions. Finally, the middleware closes any remaining open resources and then cleans up the context itself.

How the middleware works:

The context we’ve been talking about is a ContextVar. It stores a mutable container, and when your application accesses the library to obtain a session, the library operates on that container. Because the container is mutable, sessions and transactions can be closed early. The middleware then operates only on what remains open within the container.

Summary

Let’s summarize. We’ve built a great library that makes working with SQLAlchemy in asynchronous applications simple and enjoyable:

Minimal code, no duplication
Automatic commit or rollback - no need for manual management
Full support for manual session and transaction control when needed
Convenient for both CRUD operations and advanced use cases
No new syntax - pure SQLAlchemy
Framework‑agnostic
Easy to test

Use it!

Licensed under the MIT License
Open‑source code on GitHub: github.com/krylosov-aa/context-async-sqlalchemy
Documentation: krylosov-aa.github.io/context-async-sqlalchemy
Available on PyPI: pypi.org/project/context-async-sqlalchemy

I’m using this library in a real production environment - so feel free to use it in your own projects as well! Your feedback is always welcome - I’m open to improvements, refinements, and suggestions.

Don't forget to give the project a star on GitHub if you like it! Thanks!