使用 SQLAlchemy 和 Mapped 创建表时指定时间戳列类型

Question

我为表定义了 Post 类，但出现以下错误。如何在我现有的类中包含时间戳以使 postgres 为每个条目注册时间戳？

主.py

from fastapi import FastAPI, Response, status, HTTPException, Depends
from fastapi.params import Body  # for storing user sending data
from typing import Optional
from pydantic import BaseModel
from random import randrange
import psycopg2
import time
from . import model
from .database import engine, Session, get_db

model.Base.metadata.create_all(bind=engine)

from psycopg2.extras import RealDictCursor

import os

app = FastAPI()


@app.get("/sqlalchemy")
def test_posts(db: Session = Depends(get_db)):
    return {"status": "success"}

# module.py
from .database import Base
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column

# from sqlalchemy.sql.sqltypes import TIMESTAMP
from sqlalchemy.sql.expression import text
from sqlalchemy.types import TIMESTAMP


class Post(Base):
    __tablename__ = "posts"
    id: Mapped[int] = mapped_column(primary_key=True, nullable=False)
    title: Mapped[str] = mapped_column(nullable=False)
    content: Mapped[str] = mapped_column(nullable=False)
    published: Mapped[bool] = mapped_column(server_default="true", nullable=False)
    created_at: Mapped[TIMESTAMP(timezone=True)] = mapped_column(
        nullable=False, server_default=text("now()")
)

错误：

File "/home/shri/Desktop/projects/.venv/lib/python3.10/site-packages/sqlalchemy/orm/properties.py", line 791, in _init_column_for_annotation
    new_sqltype = registry._resolve_type(check_type)
  File "/home/shri/Desktop/projects/.venv/lib/python3.10/site-packages/sqlalchemy/orm/decl_api.py", line 1268, in _resolve_type
    search = ((pt, pt) for pt in python_type_type.__mro__)
AttributeError: 'TIMESTAMP' object has no attribute '__mro__'

我想知道如何正确定义；

created_at: Mapped[TIMESTAMP(timezone=True)] = mapped_column(
    nullable=False, server_default=text("now()")
)

制作 posgres 以实现自动时间注册

Answer 1

Mapped

中的类型必须是Python类型，而不是SQLAlchemy类型。这个声明应该有效：

from datetime import datetime

created_at: Mapped[datetime] = mapped_column(
    TIMESTAMP(timezone=True), server_default=sa.text('now()')
)

（请注意，您不需要提供 nullable，因为映射类型是否可选）。

另请参阅新增内容，了解 2.0 中的输入内容。