postgres-wire

This module provides a postgres wire compatible server for Python that leverages Pydantic models for brevity.

$ # The example below is being run in a separate terminal
$ psql postgresql://user:password@localhost:5432/db
psql (14.19 (Homebrew), server 0.0.0)
Type "help"for help.
db=> SELECT t;
foo | bar
--------+-----
'wow'| 12
'woah'| 21
(2 rows)
db=>

Installing

You should be able to use your preferred package manager to fetch the latest version from PyPi.

$ uv add postgres-wire
$ # or pip install postgres-wire

Usage

frompostgres_wireimportcreate_server, HandlerfrompydanticimportBaseModelclassYourObject(BaseModel):
foo: strbar: intclassYourHandler(Handler):
# If you'd like to listen on a different port# port = 55432defquery(self, sql):
print(f"Handling the query <{sql}>")
return [YourObject(foo="wow", bar=12), YourObject(foo="woah", bar=21)]
serve=create_server(YourHandler)
serve() # You can now connect with `psql postgresql://user:pass@localhost:5432`

Usage with auth

If you're interested in adding auth (ie API keys), then implement a check_auth method for the handler object which accepts a username and password as arguments.

frompostgres_wireimportcreate_server, HandlerfrompydanticimportBaseModelclassYourObject(BaseModel):
foo: strbar: intclassYourHandler(Handler):
# If you'd like to listen on a different port# port = 55432defquery(self, sql):
print(f"Handling the query <{sql}>")
return [YourObject(foo="wow", bar=12), YourObject(foo="woah", bar=21)]
defcheck_auth(self, user, password):
ifpassword=="pass":
# Raising an exception will cause the authentication to failraiseValueErrorserve=create_server(YourHandler)
serve() # You can now connect with `psql postgresql://user:password@localhost:5432`

Background

The initial code is taken from this gist which was an improvement atop this gist.