Блог

Я пытаюсь создать таблицу для обработки частоты выставления счетов с помощью ORM SQLAlchemy, и, похоже, я не могу заставить ее быть счастливой

Следующее отлично работает в Postgres:

 create table test_interval(
    frequency interval primary key
);


insert into test_interval values ('1 MONTH'), ('1 YEAR');


select * from test_interval;
-- 0 years 1 mons 0 days 0 hours 0 mins 0.00 secs
-- 1 years 0 mons 0 days 0 hours 0 mins 0.00 secs

  

И сейчас я пытаюсь добиться того же в SQLAlchemy с помощью этого кода

 from typing import Any

from sqlalchemy import Column, Interval, PrimaryKeyConstraint
from sqlalchemy.ext.declarative import as_declarative, declared_attr


@as_declarative()
class Base:
    id: Any
    __name__: str

    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()



class BillingFrequency(Base):
    __tablename__ = "billing_frequency"
    # I've also tried this
    # __table_args__ = (PrimaryKeyConstraint("frequency"),)
    # frequency: Column(Interval(native=True), index=True, unique=True, nullable=False)
    frequency: Column(Interval(native=True), primary_key=True, nullable=False)


# seed.py
# -- I've not even managed to create the table so this is yet untested --
from sqlalchemy.orm import Session
from sqlalchemy.dialects.postgresql import insert

from app.models import BillingFrequency

def seed_billing(db: Session) -> None:
    # Monthy frequency
    stmt_month = insert(BillingFrequency).values(frequency="1 MONTH")
    stmt_month = stmt_month.on_conflict_do_nothing(
        index_elements=[BillingFrequency.frequency],
    )
    db.add(stmt_month)
    # Year frequency
    stmt_year = insert(BillingFrequency).values(frequency="1 YEAR")
    stmt_year = stmt_year.on_conflict_do_nothing(
        index_elements=[BillingFrequency.frequency],
    )
    db.add(stmt_year)
    db.commit()


  

Что приводит к следующей ошибке:

 sqlalchemy.exc.ArgumentError: Mapper mapped class BillingFrequency->billing_frequency could not assemble any primary key columns for mapped table 'billing_frequency'
  

И если я попытаюсь использовать первичный ключ, используя __table_args__ , я получаю следующую ошибку.

  KeyError: 'frequency'
  

Не уверен, как с этим справиться. Это довольно тривиально в чистом SQL, но ORM делает это болезненным.

Вы допустили две незначительные ошибки, но сообщения об ошибках, к сожалению, немного загадочны для такого типа ошибок.

Первая проблема заключается в том, что вы использовали ...: Column т.е. Как тип вместо ...= Column присвоения значения. Это то, что вызывает sqlalchemy.exc.ArgumentError и KeyError: 'frequency' , SQLAlchemy не знает, что столбец существует, поскольку он не просматривает аннотации типов для данных столбца.

Вторая ошибка, которую вы допустили, заключается в использовании db.add(…) для оператора, вместо этого вы должны использовать db.execute(…) . Вы получите следующую ошибку с db.add :

 AttributeError: 'Insert' object has no attribute '_sa_instance_state'

The above exception was the direct cause of the following exception:
...
sqlalchemy.orm.exc.UnmappedInstanceError: Class 'sqlalchemy.dialects.postgresql.dml.Insert' is not mapped

  

С этими изменениями ваш код должен выглядеть следующим образом:

 
from typing import Any

from sqlalchemy import Column, Interval, PrimaryKeyConstraint
from sqlalchemy.ext.declarative import as_declarative, declared_attr


@as_declarative()
class Base:
    id: Any
    __name__: str

    # Generate __tablename__ automatically
    @declared_attr
    def __tablename__(cls) -> str:
        return cls.__name__.lower()



class BillingFrequency(Base):
    __tablename__ = "billing_frequency"
    frequency = Column(Interval(native=True), primary_key=True, nullable=False)


# seed.py
# -- I've not even managed to create the table so this is yet untested --
from sqlalchemy.orm import Session
from sqlalchemy.dialects.postgresql import insert

from app.models import BillingFrequency

def seed_billing(db: Session) -> None:
    # Monthy frequency
    stmt_month = insert(BillingFrequency).values(frequency="1 MONTH")
    stmt_month = stmt_month.on_conflict_do_nothing(
        index_elements=[BillingFrequency.frequency],
    )
    db.execute(stmt_month)
    # Year frequency
    stmt_year = insert(BillingFrequency).values(frequency="1 YEAR")
    stmt_year = stmt_year.on_conflict_do_nothing(
        index_elements=[BillingFrequency.frequency],
    )
    db.execute(stmt_year)
    db.commit()