can't subtract offset-naive and offset-aware datetimes

[om-livebird]

from sqlalchemy import Column, String, DateTime, ForeignKey, Enum, Integer
from sqlalchemy.orm import relationship
from sqlalchemy_history import make_versioned, versioning_manager
from app.db import Base, SoftDeleteMixin
from app.util_functions import get_utc_now
from app.v1.computed_attribute.dtos.computed_attribute_enum import OutputDataType

make_versioned(user_cls=None)


class ComputedAttribute(Base, SoftDeleteMixin):
    __tablename__ = "computed_attribute"
    __versioned__ = {}

    name = Column(String, nullable=False)
    description = Column(String, nullable=True)
    expression = Column(String, nullable=False)
    output_data_type = Column(Enum(OutputDataType), nullable=False)
    execution_index = Column(Integer, nullable=False, default=0, server_default="0")

    entity_id = Column(Integer, ForeignKey("entity.id"), nullable=True)
    group_id = Column(Integer, ForeignKey("entity_group.id"), nullable=True)

    created_at = Column(DateTime(timezone=True), nullable=False, default=get_utc_now)
    updated_at = Column(
        DateTime(timezone=True),
        nullable=False,
        default=get_utc_now,
        onupdate=get_utc_now,
    )

    entity = relationship("Entity")
    group = relationship("EntityGroup")

I have this table, in this I added versioning, After I generated the alembic migration which looks like this

def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        "computed_attribute_version",
        sa.Column("name", sa.String(), autoincrement=False, nullable=True),
        sa.Column("description", sa.String(), autoincrement=False, nullable=True),
        sa.Column("expression", sa.String(), autoincrement=False, nullable=True),
        sa.Column(
            "output_data_type",
            ENUM(
                "NUMBER", "BOOLEAN", "STRING", name="outputdatatype", create_type=False
            ),
            autoincrement=False,
            nullable=True,
        ),
        sa.Column("execution_index", sa.Integer(), autoincrement=False, nullable=True),
        sa.Column("entity_id", sa.Integer(), autoincrement=False, nullable=True),
        sa.Column("group_id", sa.Integer(), autoincrement=False, nullable=True),
        sa.Column(
            "created_at", sa.DateTime(timezone=True), autoincrement=False, nullable=True
        ),
        sa.Column(
            "updated_at", sa.DateTime(timezone=True), autoincrement=False, nullable=True
        ),
        sa.Column("id", sa.Integer(), autoincrement=False, nullable=False),
        sa.Column("is_deleted", sa.Boolean(), autoincrement=False, nullable=True),
        sa.Column(
            "transaction_id", sa.BigInteger(), autoincrement=False, nullable=False
        ),
        sa.Column("end_transaction_id", sa.BigInteger(), nullable=True),
        sa.Column("operation_type", sa.SmallInteger(), nullable=False),
        sa.PrimaryKeyConstraint("id", "transaction_id"),
    )
    op.create_index(
        op.f("ix_computed_attribute_version_end_transaction_id"),
        "computed_attribute_version",
        ["end_transaction_id"],
        unique=False,
    )
    op.create_index(
        op.f("ix_computed_attribute_version_operation_type"),
        "computed_attribute_version",
        ["operation_type"],
        unique=False,
    )
    op.create_index(
        op.f("ix_computed_attribute_version_transaction_id"),
        "computed_attribute_version",
        ["transaction_id"],
        unique=False,
    )
    op.create_table(
        "transaction",
        sa.Column("id", sa.BigInteger(), autoincrement=True, nullable=False),
        sa.Column("remote_addr", sa.String(length=50), nullable=True),
        sa.Column("issued_at", sa.DateTime(), nullable=True),
        sa.PrimaryKeyConstraint("id"),
    )
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_table("transaction")
    op.drop_index(
        op.f("ix_computed_attribute_version_transaction_id"),
        table_name="computed_attribute_version",
    )
    op.drop_index(
        op.f("ix_computed_attribute_version_operation_type"),
        table_name="computed_attribute_version",
    )
    op.drop_index(
        op.f("ix_computed_attribute_version_end_transaction_id"),
        table_name="computed_attribute_version",
    )
    op.drop_table("computed_attribute_version")
    # ### end Alembic commands ###

But now when I add a record in DB, I get the error

[SQL: INSERT INTO transaction (id, remote_addr, issued_at) VALUES (nextval('transaction_id_seq'), $1::VARCHAR, $2::TIMESTAMP WITHOUT TIME ZONE) RETURNING transaction.id]
[parameters: (None, datetime.datetime(2025, 12, 8, 11, 44, 22, 254552, tzinfo=datetime.timezone.utc))]
(Background on this error at: https://sqlalche.me/e/20/dbapi)

What wrong am I doing?

[om-livebird]

Note: I have also added

configure_mappers()

in env.py file of alembic

[AbdealiLoKo]

Hi @om-livebird to help here, we would need more info... I cannot see why the error comes up based on the info provided so far.

Question 1 - Currently there seem to be a lot of classes in the given code like SoftDeleteMixin which are in your application. Is the application open source ? or can you provide a MRE ?

Question 2 - Which DB are you using (and version) ?

Question 3 - Do you have any custom sqlalchemy compilation hooks for sa.DateTime() or something similar ?

[om-livebird]

Hi, thank you for looking into this! Here are the answers to your questions:

Question 1 - SoftDeleteMixin / MRE

The application is not open source, but SoftDeleteMixin is a simple mixin that just adds a boolean column:

from sqlalchemy import Column, Boolean

class SoftDeleteMixin:
    is_deleted = Column(Boolean, nullable=False, default=False, name="is_deleted")

Here's a minimal reproducible example (complete standalone script):

"""
MRE for sqlalchemy-history transaction table insertion error
Run this script to reproduce the issue.
Requires: pip install sqlalchemy sqlalchemy-history asyncpg alembic
"""

import asyncio
from sqlalchemy import Column, String, DateTime, Enum, Integer, Boolean
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import configure_mappers
from sqlalchemy_history import make_versioned
from datetime import datetime, timezone
import enum

# Initialize versioning
make_versioned(user_cls=None)

Base = declarative_base()


def get_utc_now():
    return datetime.now(timezone.utc)


class ProductType(enum.Enum):
    ELECTRONICS = "ELECTRONICS"
    CLOTHING = "CLOTHING"
    FOOD = "FOOD"


class SoftDeleteMixin:
    is_deleted = Column(Boolean, nullable=False, default=False, name="is_deleted")


class Product(Base, SoftDeleteMixin):
    __tablename__ = "product"
    __versioned__ = {}

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    description = Column(String, nullable=True)
    sku = Column(String, nullable=False)
    product_type = Column(Enum(ProductType), nullable=False)
    price = Column(Integer, nullable=False, default=0, server_default="0")

    category_id = Column(Integer, nullable=True)
    supplier_id = Column(Integer, nullable=True)

    created_at = Column(DateTime(timezone=True), nullable=False, default=get_utc_now)
    updated_at = Column(
        DateTime(timezone=True),
        nullable=False,
        default=get_utc_now,
        onupdate=get_utc_now,
    )


async def main():
    # Update this with your PostgreSQL connection string (use asyncpg driver)
    DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/testdb"

    engine = create_async_engine(DATABASE_URL, echo=True)
    AsyncSessionLocal = async_sessionmaker(
        bind=engine, class_=AsyncSession, expire_on_commit=False
    )

    # Create all tables (including version tables)
    configure_mappers()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)

    # Create a session and try to insert a record
    async with AsyncSessionLocal() as session:
        try:
            new_record = Product(
                name="Sample Product",
                description="Test Description",
                sku="SKU-001",
                product_type=ProductType.ELECTRONICS,
                price=100,
            )
            session.add(new_record)
            await session.commit()
            print("Record inserted successfully!")
        except Exception as e:
            print(f"Error occurred: {e}")
            await session.rollback()
    await engine.dispose()


if __name__ == "__main__":
    asyncio.run(main())

Question 2 - Database and Version

• Database: PostgreSQL
• PostgreSQL Version: PostgreSQL 17.6 on x86_64-windows, compiled by msvc-19.44.35213, 64-bit
• Python packages:
  • SQLAlchemy==2.0.41
  • asyncpg==0.30.0 (async driver)
  • psycopg2-binary==2.9.10 (sync driver for migrations)
  • alembic==1.13.3

Question 3 - Custom SQLAlchemy compilation hooks for DateTime

No, I do not have any custom compilation hooks (@compiles), TypeDecorators, or similar customizations for sa.DateTime().

The only DateTime-related utility is a simple function:

from datetime import datetime, timezone

def get_utc_now() -> datetime:
    return datetime.now(timezone.utc)