Connection

The DBConnection class manages all Postgres-Python interactions that actually affect the state of the database.

CLASSiceaxe.session.DBConnection

DBConnection

Core class for all ORM actions against a PostgreSQL database. Provides high-level methods for executing queries and managing database transactions.

The DBConnection wraps an asyncpg Connection and provides ORM functionality for:

  • Executing SELECT/INSERT/UPDATE/DELETE queries
  • Managing transactions
  • Inserting, updating, and deleting model instances
  • Refreshing model instances from the database

Class Constructor

  • Name
    conn
    Type
    asyncpg.Connection
    Description

    An asyncpg Connection instance to wrap

Class Attributes

  • Name
    conn
    Type
    Description
  • Name
    obj_to_primary_key
    Type
    dict[str, str | None]
    Description
  • Name
    in_transaction
    Type
    Description

Class Methods

  • Name
    transaction
    Return type
    Description

    Context manager for managing database transactions. Ensures that a series of database operations are executed atomically.

  • Name
    exec
    Return type
    list[T] | None
    Description

    Execute a query built with QueryBuilder and return the results.

  • Name
    insert
    Return type
    Description

    Insert one or more model instances into the database. If the model has an auto-incrementing primary key, it will be populated on the instances after insertion.

  • Name
    upsert
    Return type
    list[tuple[T, *Ts]] | None
    Description

    Performs an upsert (INSERT ... ON CONFLICT DO UPDATE) operation for the given objects. This is useful when you want to insert records but update them if they already exist.

  • Name
    update
    Return type
    Description

    Update one or more model instances in the database. Only modified attributes will be updated.

  • Name
    delete
    Return type
    Description

    Delete one or more model instances from the database.

  • Name
    refresh
    Return type
    Description

    Refresh one or more model instances from the database, updating their attributes with the current database values.

  • Name
    get
    Return type
    TableType | None
    Description

    Retrieve a single model instance by its primary key value.

    This method provides a convenient way to fetch a single record from the database using its primary key. It automatically constructs and executes a SELECT query with a WHERE clause matching the primary key.

Code
# Create a connection
conn = DBConnection(
    await asyncpg.connect(
        host="localhost",
        port=5432,
        user="db_user",
        password="yoursecretpassword",
        database="your_db",
    )
)

# Use with models
class User(TableBase):
    id: int = Field(primary_key=True)
    name: str
    email: str

# Insert data
user = User(name="Alice", email="alice@example.com")
await conn.insert([user])

# Query data
users = await conn.exec(
    select(User)
    .where(User.name == "Alice")
)

# Update data
user.email = "newemail@example.com"
await conn.update([user])