Extensions
You can use extensions that require extending the driver by for e.g. adding additional types. If the extension you're
interested in provides such an extension compatible with psycopg AsyncConnection
, you can use it as you would with
psycopg directly. You can pass these extensions to the patch
parameter when instantiating your instance of
Postgres
.
pg = Postgres(
"postgres", "password", "localhost", patch=[register_my_extension]
)
Example: pgvector
import numpy as np
from pydantic import BaseModel, ConfigDict
from pgvector.psycopg import register_vector_async
class Item(BaseModel):
model_config = ConfigDict(arbitrary_types_allowed=True)
id: int
embedding: np.ndarray
pg = Postgres(
"postgres", "password", "localhost", patch=[register_vector_async]
)
async def main():
await pg(
"CREATE TABLE items (id bigserial PRIMARY KEY, embedding vector(3));"
)
embedding = np.array([1, 2, 3])
await pg("INSERT INTO items (embedding) VALUES (%s)", (embedding,))
res = await pg(
"SELECT * FROM items ORDER BY embedding <=> %s;",
(embedding,),
model=Item,
)
print(res)
[Item(id=1, embedding=array([1., 2., 3.], dtype=float32))]