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

  • Name
    uncommitted_verbosity
    Type
    Literal['ERROR', 'WARNING', 'INFO'] | None
    Description

    Default: None

    The verbosity level if objects are modified but not committed when the session is closed, defaults to nothing

Class Attributes

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

Class Methods

  • Name
    initialize_types
    Return type
    None
    Description

    Introspect and register PostgreSQL type codecs on this connection, caching the result globally using the connection's DB URL as a key. These types are unlikely to change in the lifetime of a Python process, so this is typically safe to do automatically.

    This method should be called once per connection so we can leverage our own cache. If asyncpg is called directly on a new connection, it will result in its own duplicate type introspection call.

  • Name
    get_dsn
    Return type
    str
    Description

    Get the DSN (Data Source Name) string for this connection.

  • 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] | TableType | 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. Updates are batched together by grouping objects with the same modified fields, then using executemany() for efficiency.

  • 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.

  • Name
    close
    Return type
    Description

    Close the database connection.

  • Name
    magic_migrate
    Return type
    None
    Description

    Automatically generate and apply migrations to sync the database schema with the current models.

    This method combines migration generation and application into a single call. It will:

    1. Generate a new migration file if schema changes are detected
    2. Apply all pending migrations (including any just generated)
    3. Track the active revision in the migration_info table
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])

Common Exceptions

DBConnection.exec() adds a small amount of ORM-specific error handling on top of asyncpg:

  • IceaxeQueryError wraps failed queries with the SQL text and variables used for execution
  • NoObjectFound is raised by .one() queries when the database returns no rows
CLASSiceaxe.exceptions.IceaxeQueryError

IceaxeQueryError

Query error that subclasses the original asyncpg exception type, adding iceaxe context (SQL text and variables) to the message.

Class Constructor

  • Name
    original
    Type
    asyncpg.PostgresError
    Description
  • Name
    sql_text
    Type
    str
    Description
  • Name
    variables
    Type
    tuple
    Description

Class Attributes

  • Name
    sqlstate
    Type
    Description
CLASSiceaxe.exceptions.NoObjectFound

NoObjectFound

Query completed successfully but returned no rows for a .one() lookup.

Class Constructor

  • Name
    object_type
    Type
    type[Any]
    Description
  • Name
    sql_text
    Type
    str
    Description
  • Name
    variables
    Type
    tuple[Any, ...]
    Description