Python ORM-JXin的博客

学习笔记说明

这是一篇Python的学习笔记，关于Python ORM，仅做记录

ORM映射
表结构同步
认识核心概念

[!NOTE]

小贴士

快速删除所有表
DROP SCHEMA public CASCADE;
CREATE SCHEMA public;
1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

1
2

ORM映射

ORM（Object Relational Mapping） 将数据库中的"表"映射为 Python 中的"类"，将"一行记录"映射为"对象实例"。

先在 model/ 下新建 base.py：

# model/base.py
from datetime import datetime

from sqlalchemy import DateTime, Identity, func
from sqlalchemy.orm import DeclarativeBase, Mapped, declared_attr, mapped_column

class Base(DeclarativeBase):
    @declared_attr.directive
    def __tablename__(cls) -> str:
        return cls.__name__.lower()

class IDMixin:
    id: Mapped[int] = mapped_column(Identity(), primary_key=True)

class TimestampMixin:
    created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
    updated_at: Mapped[datetime] = mapped_column(
        DateTime, server_default=func.now(), onupdate=func.now()
    )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

所有模型类继承这个 Base。

Category 模型

# model/category.py
from typing import TYPE_CHECKING

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.model.association import product_category
from app.model.base import Base, IDMixin

if TYPE_CHECKING:
    from app.model.product import Product

class Category(Base, IDMixin):

    name: Mapped[str] = mapped_column(String(50))
    description: Mapped[str] = mapped_column(Text, default="")

    products: Mapped[list["Product"]] = relationship(
        secondary=product_category, back_populates="categories"
    )

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

relationship 此时只是声明关系，不影响建表。建表只认 mapped_column。

Product 模型

# model/product.py
from typing import TYPE_CHECKING

from sqlalchemy import String, Text
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.model.association import product_category
from app.model.base import Base, IDMixin, TimestampMixin

if TYPE_CHECKING:
    from app.model.category import Category
    from app.model.sku import Sku

class Product(Base, IDMixin, TimestampMixin):

    name: Mapped[str] = mapped_column(String(200))
    description: Mapped[str] = mapped_column(Text, default="")
    brand: Mapped[str | None] = mapped_column(String(100))

    categories: Mapped[list["Category"]] = relationship(
        secondary=product_category, back_populates="products"
    )
    skus: Mapped[list["Sku"]] = relationship(back_populates="product")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

SKU 模型

# model/sku.py
from decimal import Decimal
from typing import TYPE_CHECKING

from sqlalchemy import String, Numeric, ForeignKey
from sqlalchemy.dialects.postgresql import JSONB
from sqlalchemy.orm import Mapped, mapped_column, relationship

from app.model.base import Base, IDMixin

if TYPE_CHECKING:
    from app.model.product import Product

class Sku(Base, IDMixin):

    product_id: Mapped[int] = mapped_column(
        ForeignKey("product.id", ondelete="CASCADE")
    )
    sku_code: Mapped[str] = mapped_column(String(50), unique=True)
    price: Mapped[Decimal] = mapped_column(Numeric(10, 2))
    stock: Mapped[int] = mapped_column(default=0)
    attrs: Mapped[dict] = mapped_column(JSONB)
    image_url: Mapped[str] = mapped_column(String)

    product: Mapped["Product"] = relationship(back_populates="skus")

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

product_category关系表

# model/association/product_category.py
from sqlalchemy import Column, ForeignKey, Table

from app.model.base import Base

product_category = Table(
    "product_category",
    Base.metadata,
    Column("product_id", ForeignKey("product.id"), primary_key=True),
    Column("category_id", ForeignKey("category.id"), primary_key=True),
)

1
2
3
4
5
6
7
8
9
10
11

# model/association/__init__.py
from app.model.association.product_category import product_category

__all__ = ["product_category"]

1
2
3
4

表结构同步

在 model/main.py 中写建表逻辑：

# model/main.py
import asyncio
from sqlalchemy import text
from app.model.engine import get_engine
from app.model.base import Base
from app.model.category import Category   # noqa: F401
from app.model.product import Product     # noqa: F401
from app.model.sku import Sku             # noqa: F401

async def init_db():
    engine = get_engine()
    async with engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)
    print("✅ 数据库表创建完成")

async def test_connection():
    engine = get_engine()
    try:
        async with engine.connect() as conn:
            result = await conn.execute(text("SELECT 1"))
            value = result.scalar()
            print(f"✅ 连接成功！查询结果: {value}")
    except Exception as e:
        print(f"❌ 连接失败: {e}")

async def main():
    await init_db()
    await test_connection()

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

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

认识核心概念

ORM