🗄 (🔗) 💽

FastAPI 🚫 🚚 👆 ⚙️ 🗄 (🔗) 💽.

✋️ 👆 💪 ⚙️ 🙆 🔗 💽 👈 👆 💚.

📥 👥 🔜 👀 🖼 ⚙️ 🇸🇲.

👆 💪 💪 🛠️ ⚫️ 🙆 💽 🐕‍🦺 🇸🇲, 💖:

• ✳
• ✳
• 🗄
• 🐸
• 🤸‍♂ 🗄 💽, ♒️.

👉 🖼, 👥 🔜 ⚙️ 🗄, ↩️ ⚫️ ⚙️ 👁 📁 & 🐍 ✔️ 🛠️ 🐕‍🦺. , 👆 💪 📁 👉 🖼 & 🏃 ⚫️.

⏪, 👆 🏭 🈸, 👆 💪 💚 ⚙️ 💽 💽 💖 ✳.

Tip

📤 🛂 🏗 🚂 ⏮️ FastAPI & ✳, 🌐 ⚓️ 🔛 ☁, 🔌 🕸 & 🌖 🧰: https://github.com/tiangolo/full-stack-fastapi-postgresql

Note

👀 👈 📚 📟 🐩 SQLAlchemy 📟 👆 🔜 ⚙️ ⏮️ 🙆 🛠️.

FastAPI 🎯 📟 🤪 🕧.

🐜

FastAPI 👷 ⏮️ 🙆 💽 & 🙆 👗 🗃 💬 💽.

⚠ ⚓ ⚙️ "🐜": "🎚-🔗 🗺" 🗃.

🐜 ✔️ 🧰 🗜 ("🗺") 🖖 🎚 📟 & 💽 🏓 ("🔗").

⏮️ 🐜, 👆 🛎 ✍ 🎓 👈 🎨 🏓 🗄 💽, 🔠 🔢 🎓 🎨 🏓, ⏮️ 📛 & 🆎.

🖼 🎓 Pet 💪 🎨 🗄 🏓 pets.

& 🔠 👐 🎚 👈 🎓 🎨 ⏭ 💽.

🖼 🎚 orion_cat (👐 Pet) 💪 ✔️ 🔢 orion_cat.type, 🏓 type. & 💲 👈 🔢 💪, ✅ "cat".

👫 🐜 ✔️ 🧰 ⚒ 🔗 ⚖️ 🔗 🖖 🏓 ⚖️ 👨‍💼.

👉 🌌, 👆 💪 ✔️ 🔢 orion_cat.owner & 👨‍💼 🔜 🔌 💽 👉 🐶 👨‍💼, ✊ ⚪️➡️ 🏓 👨‍💼.

, orion_cat.owner.name 💪 📛 (⚪️➡️ name 🏓 owners 🏓) 👉 🐶 👨‍💼.

⚫️ 💪 ✔️ 💲 💖 "Arquilian".

& 🐜 🔜 🌐 👷 🤚 ℹ ⚪️➡️ 🔗 🏓 👨‍💼 🕐❔ 👆 🔄 🔐 ⚫️ ⚪️➡️ 👆 🐶 🎚.

⚠ 🐜 🖼: ✳-🐜 (🍕 ✳ 🛠️), 🇸🇲 🐜 (🍕 🇸🇲, 🔬 🛠️) & 🏒 (🔬 🛠️), 👪 🎏.

📥 👥 🔜 👀 ❔ 👷 ⏮️ 🇸🇲 🐜.

🎏 🌌 👆 💪 ⚙️ 🙆 🎏 🐜.

Tip

📤 🌓 📄 ⚙️ 🏒 📥 🩺.

📁 📊

👫 🖼, ➡️ 💬 👆 ✔️ 📁 📛 my_super_project 👈 🔌 🎧-📁 🤙 sql_app ⏮️ 📊 💖 👉:

.
└── sql_app
    ├── __init__.py
    ├── crud.py
    ├── database.py
    ├── main.py
    ├── models.py
    └── schemas.py

📁 __init__.py 🛁 📁, ✋️ ⚫️ 💬 🐍 👈 sql_app ⏮️ 🌐 🚮 🕹 (🐍 📁) 📦.

🔜 ➡️ 👀 ⚫️❔ 🔠 📁/🕹 🔨.

❎ SQLAlchemy

🥇 👆 💪 ❎ SQLAlchemy:

$ pip install sqlalchemy

---> 100%

✍ 🇸🇲 🍕

➡️ 🔗 📁 sql_app/database.py.

🗄 🇸🇲 🍕

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

✍ 💽 📛 🇸🇲

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

👉 🖼, 👥 "🔗" 🗄 💽 (📂 📁 ⏮️ 🗄 💽).

📁 🔜 🔎 🎏 📁 📁 sql_app.db.

👈 ⚫️❔ 🏁 🍕 ./sql_app.db.

🚥 👆 ⚙️ ✳ 💽 ↩️, 👆 🔜 ✔️ ✍ ⏸:

SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

...& 🛠️ ⚫️ ⏮️ 👆 💽 📊 & 🎓 (📊 ✳, ✳ ⚖️ 🙆 🎏).

Tip

👉 👑 ⏸ 👈 👆 🔜 ✔️ 🔀 🚥 👆 💚 ⚙️ 🎏 💽.

✍ 🇸🇲 engine

🥇 🔁 ✍ 🇸🇲 "🚒".

👥 🔜 ⏪ ⚙️ 👉 engine 🎏 🥉.

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

🗒

❌:

connect_args={"check_same_thread": False}

...💪 🕴 SQLite. ⚫️ 🚫 💪 🎏 💽.

📡 ℹ

🔢 🗄 🔜 🕴 ✔ 1️⃣ 🧵 🔗 ⏮️ ⚫️, 🤔 👈 🔠 🧵 🔜 🍵 🔬 📨.

👉 ❎ 😫 🤝 🎏 🔗 🎏 👜 (🎏 📨).

✋️ FastAPI, ⚙️ 😐 🔢 (def) 🌅 🌘 1️⃣ 🧵 💪 🔗 ⏮️ 💽 🎏 📨, 👥 💪 ⚒ 🗄 💭 👈 ⚫️ 🔜 ✔ 👈 ⏮️ connect_args={"check_same_thread": False}.

, 👥 🔜 ⚒ 💭 🔠 📨 🤚 🚮 👍 💽 🔗 🎉 🔗, 📤 🙅‍♂ 💪 👈 🔢 🛠️.

✍ SessionLocal 🎓

🔠 👐 SessionLocal 🎓 🔜 💽 🎉. 🎓 ⚫️ 🚫 💽 🎉.

✋️ 🕐 👥 ✍ 👐 SessionLocal 🎓, 👉 👐 🔜 ☑ 💽 🎉.

👥 📛 ⚫️ SessionLocal 🔬 ⚫️ ⚪️➡️ Session 👥 🏭 ⚪️➡️ 🇸🇲.

👥 🔜 ⚙️ Session (1️⃣ 🗄 ⚪️➡️ 🇸🇲) ⏪.

✍ SessionLocal 🎓, ⚙️ 🔢 sessionmaker:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

✍ Base 🎓

🔜 👥 🔜 ⚙️ 🔢 declarative_base() 👈 📨 🎓.

⏪ 👥 🔜 😖 ⚪️➡️ 👉 🎓 ✍ 🔠 💽 🏷 ⚖️ 🎓 (🐜 🏷):

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

✍ 💽 🏷

➡️ 🔜 👀 📁 sql_app/models.py.

✍ 🇸🇲 🏷 ⚪️➡️ Base 🎓

👥 🔜 ⚙️ 👉 Base 🎓 👥 ✍ ⏭ ✍ 🇸🇲 🏷.

Tip

🇸🇲 ⚙️ ⚖ "🏷" 🔗 👉 🎓 & 👐 👈 🔗 ⏮️ 💽.

✋️ Pydantic ⚙️ ⚖ "🏷" 🔗 🕳 🎏, 💽 🔬, 🛠️, & 🧾 🎓 & 👐.

🗄 Base ⚪️➡️ database (📁 database.py ⚪️➡️ 🔛).

✍ 🎓 👈 😖 ⚪️➡️ ⚫️.

👫 🎓 🇸🇲 🏷.

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

__tablename__ 🔢 💬 🇸🇲 📛 🏓 ⚙️ 💽 🔠 👫 🏷.

✍ 🏷 🔢/🏓

🔜 ✍ 🌐 🏷 (🎓) 🔢.

🔠 👫 🔢 🎨 🏓 🚮 🔗 💽 🏓.

👥 ⚙️ Column ⚪️➡️ 🇸🇲 🔢 💲.

& 👥 🚶‍♀️ 🇸🇲 🎓 "🆎", Integer, String, & Boolean, 👈 🔬 🆎 💽, ❌.

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

✍ 💛

🔜 ✍ 💛.

👉, 👥 ⚙️ relationship 🚚 🇸🇲 🐜.

👉 🔜 ▶️️, 🌅 ⚖️ 🌘, "🎱" 🔢 👈 🔜 🔌 💲 ⚪️➡️ 🎏 🏓 🔗 👉 1️⃣.

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

🕐❔ 🔐 🔢 items User, my_user.items, ⚫️ 🔜 ✔️ 📇 Item 🇸🇲 🏷 (⚪️➡️ items 🏓) 👈 ✔️ 💱 🔑 ☝ 👉 ⏺ users 🏓.

🕐❔ 👆 🔐 my_user.items, 🇸🇲 🔜 🤙 🚶 & ☕ 🏬 ⚪️➡️ 💽 items 🏓 & 🔗 👫 📥.

& 🕐❔ 🔐 🔢 owner Item, ⚫️ 🔜 🔌 User 🇸🇲 🏷 ⚪️➡️ users 🏓. ⚫️ 🔜 ⚙️ owner_id 🔢/🏓 ⏮️ 🚮 💱 🔑 💭 ❔ ⏺ 🤚 ⚪️➡️ users 🏓.

✍ Pydantic 🏷

🔜 ➡️ ✅ 📁 sql_app/schemas.py.

Tip

❎ 😨 🖖 🇸🇲 🏷 & Pydantic 🏷, 👥 🔜 ✔️ 📁 models.py ⏮️ 🇸🇲 🏷, & 📁 schemas.py ⏮️ Pydantic 🏷.

👫 Pydantic 🏷 🔬 🌅 ⚖️ 🌘 "🔗" (☑ 📊 💠).

👉 🔜 ℹ 👥 ❎ 😨 ⏪ ⚙️ 👯‍♂️.

✍ ▶️ Pydantic 🏷 / 🔗

✍ ItemBase & UserBase Pydantic 🏷 (⚖️ ➡️ 💬 "🔗") ✔️ ⚠ 🔢 ⏪ 🏗 ⚖️ 👂 📊.

& ✍ ItemCreate & UserCreate 👈 😖 ⚪️➡️ 👫 (👫 🔜 ✔️ 🎏 🔢), ➕ 🙆 🌖 📊 (🔢) 💪 🏗.

, 👩‍💻 🔜 ✔️ password 🕐❔ 🏗 ⚫️.

✋️ 💂‍♂, password 🏆 🚫 🎏 Pydantic 🏷, 🖼, ⚫️ 🏆 🚫 📨 ⚪️➡️ 🛠️ 🕐❔ 👂 👩‍💻.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛🐍 3️⃣.1️⃣0️⃣ & 🔛

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

🇸🇲 👗 & Pydantic 👗

👀 👈 🇸🇲 🏷 🔬 🔢 ⚙️ =, & 🚶‍♀️ 🆎 🔢 Column, 💖:

name = Column(String)

⏪ Pydantic 🏷 📣 🆎 ⚙️ :, 🆕 🆎 ✍ ❕/🆎 🔑:

name: str

✔️ ⚫️ 🤯, 👆 🚫 🤚 😕 🕐❔ ⚙️ = & : ⏮️ 👫.

✍ Pydantic 🏷 / 🔗 👂 / 📨

🔜 ✍ Pydantic 🏷 (🔗) 👈 🔜 ⚙️ 🕐❔ 👂 💽, 🕐❔ 🛬 ⚫️ ⚪️➡️ 🛠️.

🖼, ⏭ 🏗 🏬, 👥 🚫 💭 ⚫️❔ 🔜 🆔 🛠️ ⚫️, ✋️ 🕐❔ 👂 ⚫️ (🕐❔ 🛬 ⚫️ ⚪️➡️ 🛠️) 👥 🔜 ⏪ 💭 🚮 🆔.

🎏 🌌, 🕐❔ 👂 👩‍💻, 👥 💪 🔜 📣 👈 items 🔜 🔌 🏬 👈 💭 👉 👩‍💻.

🚫 🕴 🆔 📚 🏬, ✋️ 🌐 💽 👈 👥 🔬 Pydantic 🏷 👂 🏬: Item.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛🐍 3️⃣.1️⃣0️⃣ & 🔛

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

Tip

👀 👈 User, Pydantic 🏷 👈 🔜 ⚙️ 🕐❔ 👂 👩‍💻 (🛬 ⚫️ ⚪️➡️ 🛠️) 🚫 🔌 password.

⚙️ Pydantic orm_mode

🔜, Pydantic 🏷 👂, Item & User, 🚮 🔗 Config 🎓.

👉 Config 🎓 ⚙️ 🚚 📳 Pydantic.

Config 🎓, ⚒ 🔢 orm_mode = True.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛🐍 3️⃣.1️⃣0️⃣ & 🔛

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

Tip

👀 ⚫️ ⚖ 💲 ⏮️ =, 💖:

orm_mode = True

⚫️ 🚫 ⚙️ : 🆎 📄 ⏭.

👉 ⚒ 📁 💲, 🚫 📣 🆎.

Pydantic orm_mode 🔜 💬 Pydantic 🏷 ✍ 💽 🚥 ⚫️ 🚫 dict, ✋️ 🐜 🏷 (⚖️ 🙆 🎏 ❌ 🎚 ⏮️ 🔢).

👉 🌌, ↩️ 🕴 🔄 🤚 id 💲 ⚪️➡️ dict,:

id = data["id"]

⚫️ 🔜 🔄 🤚 ⚫️ ⚪️➡️ 🔢,:

id = data.id

& ⏮️ 👉, Pydantic 🏷 🔗 ⏮️ 🐜, & 👆 💪 📣 ⚫️ response_model ❌ 👆 ➡ 🛠️.

👆 🔜 💪 📨 💽 🏷 & ⚫️ 🔜 ✍ 💽 ⚪️➡️ ⚫️.

📡 ℹ 🔃 🐜 📳

🇸🇲 & 📚 🎏 🔢 "🙃 🚚".

👈 ⛓, 🖼, 👈 👫 🚫 ☕ 💽 💛 ⚪️➡️ 💽 🚥 👆 🔄 🔐 🔢 👈 🔜 🔌 👈 💽.

🖼, 🔐 🔢 items:

current_user.items

🔜 ⚒ 🇸🇲 🚶 items 🏓 & 🤚 🏬 👉 👩‍💻, ✋️ 🚫 ⏭.

🍵 orm_mode, 🚥 👆 📨 🇸🇲 🏷 ⚪️➡️ 👆 ➡ 🛠️, ⚫️ 🚫🔜 🔌 💛 💽.

🚥 👆 📣 📚 💛 👆 Pydantic 🏷.

✋️ ⏮️ 🐜 📳, Pydantic ⚫️ 🔜 🔄 🔐 💽 ⚫️ 💪 ⚪️➡️ 🔢 (↩️ 🤔 dict), 👆 💪 📣 🎯 💽 👆 💚 📨 & ⚫️ 🔜 💪 🚶 & 🤚 ⚫️, ⚪️➡️ 🐜.

💩 🇨🇻

🔜 ➡️ 👀 📁 sql_app/crud.py.

👉 📁 👥 🔜 ✔️ ♻ 🔢 🔗 ⏮️ 💽 💽.

💩 👟 ⚪️➡️: 🅱📧, Ⓜ💳, 👤 = , & 🇨🇮📧.

...👐 👉 🖼 👥 🕴 🏗 & 👂.

✍ 💽

🗄 Session ⚪️➡️ sqlalchemy.orm, 👉 🔜 ✔ 👆 📣 🆎 db 🔢 & ✔️ 👻 🆎 ✅ & 🛠️ 👆 🔢.

🗄 models (🇸🇲 🏷) & schemas (Pydantic 🏷 / 🔗).

✍ 🚙 🔢:

• ✍ 👁 👩‍💻 🆔 & 📧.
• ✍ 💗 👩‍💻.
• ✍ 💗 🏬.

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Tip

🏗 🔢 👈 🕴 💡 🔗 ⏮️ 💽 (🤚 👩‍💻 ⚖️ 🏬) 🔬 👆 ➡ 🛠️ 🔢, 👆 💪 🌖 💪 ♻ 👫 💗 🍕 & 🚮 ⚒ 💯 👫.

✍ 💽

🔜 ✍ 🚙 🔢 ✍ 💽.

🔁:

• ✍ 🇸🇲 🏷 👐 ⏮️ 👆 📊.
• add 👈 👐 🎚 👆 💽 🎉.
• commit 🔀 💽 (👈 👫 🖊).
• refresh 👆 👐 (👈 ⚫️ 🔌 🙆 🆕 📊 ⚪️➡️ 💽, 💖 🏗 🆔).

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

Tip

🇸🇲 🏷 User 🔌 hashed_password 👈 🔜 🔌 🔐 #️⃣ ⏬ 🔐.

✋️ ⚫️❔ 🛠️ 👩‍💻 🚚 ⏮️ 🔐, 👆 💪 ⚗ ⚫️ & 🏗 #️⃣ 🔐 👆 🈸.

& ⤴️ 🚶‍♀️ hashed_password ❌ ⏮️ 💲 🖊.

Warning

👉 🖼 🚫 🔐, 🔐 🚫#️⃣.

🎰 👨‍❤‍👨 🈸 👆 🔜 💪 #️⃣ 🔐 & 🙅 🖊 👫 🔢.

🌅 ℹ, 🚶 🔙 💂‍♂ 📄 🔰.

📥 👥 🎯 🕴 🔛 🧰 & 👨‍🔧 💽.

Tip

↩️ 🚶‍♀️ 🔠 🇨🇻 ❌ Item & 👂 🔠 1️⃣ 👫 ⚪️➡️ Pydantic 🏷, 👥 🏭 dict ⏮️ Pydantic 🏷'Ⓜ 📊 ⏮️:

item.dict()

& ⤴️ 👥 🚶‍♀️ dict'Ⓜ 🔑-💲 👫 🇨🇻 ❌ 🇸🇲 Item, ⏮️:

Item(**item.dict())

& ⤴️ 👥 🚶‍♀️ ➕ 🇨🇻 ❌ owner_id 👈 🚫 🚚 Pydantic 🏷, ⏮️:

Item(**item.dict(), owner_id=user_id)

👑 FastAPI 📱

& 🔜 📁 sql_app/main.py ➡️ 🛠️ & ⚙️ 🌐 🎏 🍕 👥 ✍ ⏭.

✍ 💽 🏓

📶 🙃 🌌 ✍ 💽 🏓:

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

⚗ 🗒

🛎 👆 🔜 🎲 🔢 👆 💽 (✍ 🏓, ♒️) ⏮️ ⚗.

& 👆 🔜 ⚙️ ⚗ "🛠️" (👈 🚮 👑 👨‍🏭).

"🛠️" ⚒ 🔁 💪 🕐❔ 👆 🔀 📊 👆 🇸🇲 🏷, 🚮 🆕 🔢, ♒️. 🔁 👈 🔀 💽, 🚮 🆕 🏓, 🆕 🏓, ♒️.

👆 💪 🔎 🖼 ⚗ FastAPI 🏗 📄 ⚪️➡️ 🏗 ⚡ - 📄. 🎯 alembic 📁 ℹ 📟.

✍ 🔗

🔜 ⚙️ SessionLocal 🎓 👥 ✍ sql_app/database.py 📁 ✍ 🔗.

👥 💪 ✔️ 🔬 💽 🎉/🔗 (SessionLocal) 📍 📨, ⚙️ 🎏 🎉 🔘 🌐 📨 & ⤴️ 🔐 ⚫️ ⏮️ 📨 🏁.

& ⤴️ 🆕 🎉 🔜 ✍ ⏭ 📨.

👈, 👥 🔜 ✍ 🆕 🔗 ⏮️ yield, 🔬 ⏭ 📄 🔃 🔗 ⏮️ yield.

👆 🔗 🔜 ✍ 🆕 🇸🇲 SessionLocal 👈 🔜 ⚙️ 👁 📨, & ⤴️ 🔐 ⚫️ 🕐 📨 🏁.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

Info

👥 🚮 🏗 SessionLocal() & 🚚 📨 try 🍫.

& ⤴️ 👥 🔐 ⚫️ finally 🍫.

👉 🌌 👥 ⚒ 💭 💽 🎉 🕧 📪 ⏮️ 📨. 🚥 📤 ⚠ ⏪ 🏭 📨.

✋️ 👆 💪 🚫 🤚 ➕1️⃣ ⚠ ⚪️➡️ 🚪 📟 (⏮️ yield). 👀 🌖 🔗 ⏮️ yield & HTTPException

& ⤴️, 🕐❔ ⚙️ 🔗 ➡ 🛠️ 🔢, 👥 📣 ⚫️ ⏮️ 🆎 Session 👥 🗄 🔗 ⚪️➡️ 🇸🇲.

👉 🔜 ⤴️ 🤝 👥 👍 👨‍🎨 🐕‍🦺 🔘 ➡ 🛠️ 🔢, ↩️ 👨‍🎨 🔜 💭 👈 db 🔢 🆎 Session:

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

📡 ℹ

🔢 db 🤙 🆎 SessionLocal, ✋️ 👉 🎓 (✍ ⏮️ sessionmaker()) "🗳" 🇸🇲 Session,, 👨‍🎨 🚫 🤙 💭 ⚫️❔ 👩‍🔬 🚚.

✋️ 📣 🆎 Session, 👨‍🎨 🔜 💪 💭 💪 👩‍🔬 (.add(), .query(), .commit(), ♒️) & 💪 🚚 👍 🐕‍🦺 (💖 🛠️). 🆎 📄 🚫 📉 ☑ 🎚.

✍ 👆 FastAPI ➡ 🛠️

🔜, 😒, 📥 🐩 FastAPI ➡ 🛠️ 📟.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

👥 🏗 💽 🎉 ⏭ 🔠 📨 🔗 ⏮️ yield, & ⤴️ 📪 ⚫️ ⏮️.

& ⤴️ 👥 💪 ✍ 🚚 🔗 ➡ 🛠️ 🔢, 🤚 👈 🎉 🔗.

⏮️ 👈, 👥 💪 🤙 crud.get_user 🔗 ⚪️➡️ 🔘 ➡ 🛠️ 🔢 & ⚙️ 👈 🎉.

Tip

👀 👈 💲 👆 📨 🇸🇲 🏷, ⚖️ 📇 🇸🇲 🏷.

✋️ 🌐 ➡ 🛠️ ✔️ response_model ⏮️ Pydantic 🏷 / 🔗 ⚙️ orm_mode, 💽 📣 👆 Pydantic 🏷 🔜 ⚗ ⚪️➡️ 👫 & 📨 👩‍💻, ⏮️ 🌐 😐 ⛽ & 🔬.

Tip

👀 👈 📤 response_models 👈 ✔️ 🐩 🐍 🆎 💖 List[schemas.Item].

✋️ 🎚/🔢 👈 List Pydantic 🏷 ⏮️ orm_mode, 💽 🔜 🗃 & 📨 👩‍💻 🛎, 🍵 ⚠.

🔃 def 🆚 async def

📥 👥 ⚙️ 🇸🇲 📟 🔘 ➡ 🛠️ 🔢 & 🔗, &, 🔄, ⚫️ 🔜 🚶 & 🔗 ⏮️ 🔢 💽.

👈 💪 ⚠ 🚚 "⌛".

✋️ 🇸🇲 🚫 ✔️ 🔗 ⚙️ await 🔗, 🔜 ⏮️ 🕳 💖:

user = await db.query(User).first()

...& ↩️ 👥 ⚙️:

user = db.query(User).first()

⤴️ 👥 🔜 📣 ➡ 🛠️ 🔢 & 🔗 🍵 async def, ⏮️ 😐 def,:

@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    ...

Info

🚥 👆 💪 🔗 👆 🔗 💽 🔁, 👀 🔁 🗄 (🔗) 💽.

📶 📡 ℹ

🚥 👆 😟 & ✔️ ⏬ 📡 💡, 👆 💪 ✅ 📶 📡 ℹ ❔ 👉 async def 🆚 def 🍵 🔁 🩺.

🛠️

↩️ 👥 ⚙️ 🇸🇲 🔗 & 👥 🚫 🚚 🙆 😇 🔌-⚫️ 👷 ⏮️ FastAPI, 👥 💪 🛠️ 💽 🛠️ ⏮️ ⚗ 🔗.

& 📟 🔗 🇸🇲 & 🇸🇲 🏷 🖖 🎏 🔬 📁, 👆 🔜 💪 🎭 🛠️ ⏮️ ⚗ 🍵 ✔️ ❎ FastAPI, Pydantic, ⚖️ 🕳 🙆.

🎏 🌌, 👆 🔜 💪 ⚙️ 🎏 🇸🇲 🏷 & 🚙 🎏 🍕 👆 📟 👈 🚫 🔗 FastAPI.

🖼, 🖥 📋 👨‍🏭 ⏮️ 🥒, 🅿, ⚖️ 📶.

📄 🌐 📁

💭 👆 🔜 ✔️ 📁 📛 my_super_project 👈 🔌 🎧-📁 🤙 sql_app.

sql_app 🔜 ✔️ 📄 📁:

• sql_app/__init__.py: 🛁 📁.

• sql_app/database.py:

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"

engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

• sql_app/models.py:

from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship

from .database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)

    items = relationship("Item", back_populates="owner")


class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True)
    title = Column(String, index=True)
    description = Column(String, index=True)
    owner_id = Column(Integer, ForeignKey("users.id"))

    owner = relationship("User", back_populates="items")

• sql_app/schemas.py:

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛🐍 3️⃣.1️⃣0️⃣ & 🔛

from typing import List, Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: List[Item] = []

    class Config:
        orm_mode = True

from typing import Union

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: Union[str, None] = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

from pydantic import BaseModel


class ItemBase(BaseModel):
    title: str
    description: str | None = None


class ItemCreate(ItemBase):
    pass


class Item(ItemBase):
    id: int
    owner_id: int

    class Config:
        orm_mode = True


class UserBase(BaseModel):
    email: str


class UserCreate(UserBase):
    password: str


class User(UserBase):
    id: int
    is_active: bool
    items: list[Item] = []

    class Config:
        orm_mode = True

• sql_app/crud.py:

from sqlalchemy.orm import Session

from . import models, schemas


def get_user(db: Session, user_id: int):
    return db.query(models.User).filter(models.User.id == user_id).first()


def get_user_by_email(db: Session, email: str):
    return db.query(models.User).filter(models.User.email == email).first()


def get_users(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.User).offset(skip).limit(limit).all()


def create_user(db: Session, user: schemas.UserCreate):
    fake_hashed_password = user.password + "notreallyhashed"
    db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user


def get_items(db: Session, skip: int = 0, limit: int = 100):
    return db.query(models.Item).offset(skip).limit(limit).all()


def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
    db_item = models.Item(**item.dict(), owner_id=user_id)
    db.add(db_item)
    db.commit()
    db.refresh(db_item)
    return db_item

• sql_app/main.py:

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


# Dependency
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

✅ ⚫️

👆 💪 📁 👉 📟 & ⚙️ ⚫️.

Info

👐, 📟 🎦 📥 🍕 💯. 🌅 📟 👉 🩺.

⤴️ 👆 💪 🏃 ⚫️ ⏮️ Uvicorn:

$ uvicorn sql_app.main:app --reload

<span style="color: green;">INFO</span>:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)

& ⤴️, 👆 💪 📂 👆 🖥 http://127.0.0.1:8000/docs.

& 👆 🔜 💪 🔗 ⏮️ 👆 FastAPI 🈸, 👂 📊 ⚪️➡️ 🎰 💽:

🔗 ⏮️ 💽 🔗

🚥 👆 💚 🔬 🗄 💽 (📁) 🔗, ➡ FastAPI, ℹ 🚮 🎚, 🚮 🏓, 🏓, ⏺, 🔀 📊, ♒️. 👆 💪 ⚙️ 💽 🖥 🗄.

⚫️ 🔜 👀 💖 👉:

👆 💪 ⚙️ 💳 🗄 🖥 💖 🗄 📋 ⚖️ ExtendsClass.

🎛 💽 🎉 ⏮️ 🛠️

🚥 👆 💪 🚫 ⚙️ 🔗 ⏮️ yield - 🖼, 🚥 👆 🚫 ⚙️ 🐍 3️⃣.7️⃣ & 💪 🚫 ❎ "🐛" 🤔 🔛 🐍 3️⃣.6️⃣ - 👆 💪 ⚒ 🆙 🎉 "🛠️" 🎏 🌌.

"🛠️" 🌖 🔢 👈 🕧 🛠️ 🔠 📨, ⏮️ 📟 🛠️ ⏭, & 📟 🛠️ ⏮️ 🔗 🔢.

✍ 🛠️

🛠️ 👥 🔜 🚮 (🔢) 🔜 ✍ 🆕 🇸🇲 SessionLocal 🔠 📨, 🚮 ⚫️ 📨 & ⤴️ 🔐 ⚫️ 🕐 📨 🏁.

🐍 3️⃣.6️⃣ & 🔛🐍 3️⃣.9️⃣ & 🔛

from typing import List

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=List[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

from fastapi import Depends, FastAPI, HTTPException, Request, Response
from sqlalchemy.orm import Session

from . import crud, models, schemas
from .database import SessionLocal, engine

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

app = FastAPI()


@app.middleware("http")
async def db_session_middleware(request: Request, call_next):
    response = Response("Internal server error", status_code=500)
    try:
        request.state.db = SessionLocal()
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response


# Dependency
def get_db(request: Request):
    return request.state.db


@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
    db_user = crud.get_user_by_email(db, email=user.email)
    if db_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    return crud.create_user(db=db, user=user)


@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    users = crud.get_users(db, skip=skip, limit=limit)
    return users


@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = crud.get_user(db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user


@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
    user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
    return crud.create_user_item(db=db, item=item, user_id=user_id)


@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    items = crud.get_items(db, skip=skip, limit=limit)
    return items

Info

👥 🚮 🏗 SessionLocal() & 🚚 📨 try 🍫.

& ⤴️ 👥 🔐 ⚫️ finally 🍫.

👉 🌌 👥 ⚒ 💭 💽 🎉 🕧 📪 ⏮️ 📨. 🚥 📤 ⚠ ⏪ 🏭 📨.

🔃 request.state

request.state 🏠 🔠 Request 🎚. ⚫️ 📤 🏪 ❌ 🎚 📎 📨 ⚫️, 💖 💽 🎉 👉 💼. 👆 💪 ✍ 🌅 🔃 ⚫️ 💃 🩺 🔃 Request 🇵🇸.

👥 👉 💼, ⚫️ ℹ 👥 🚚 👁 💽 🎉 ⚙️ 🔘 🌐 📨, & ⤴️ 🔐 ⏮️ (🛠️).

🔗 ⏮️ yield ⚖️ 🛠️

❎ 🛠️ 📥 🎏 ⚫️❔ 🔗 ⏮️ yield 🔨, ⏮️ 🔺:

• ⚫️ 🚚 🌖 📟 & 👄 🌅 🏗.
• 🛠️ ✔️ async 🔢.
  • 🚥 📤 📟 ⚫️ 👈 ✔️ "⌛" 🕸, ⚫️ 💪 "🍫" 👆 🈸 📤 & 📉 🎭 🍖.
  • 👐 ⚫️ 🎲 🚫 📶 ⚠ 📥 ⏮️ 🌌 SQLAlchemy 👷.
  • ✋️ 🚥 👆 🚮 🌖 📟 🛠️ 👈 ✔️ 📚 👤/🅾 ⌛, ⚫️ 💪 ⤴️ ⚠.
• 🛠️ 🏃 🔠 📨.
  • , 🔗 🔜 ✍ 🔠 📨.
  • 🕐❔ ➡ 🛠️ 👈 🍵 👈 📨 🚫 💪 💽.

Tip

⚫️ 🎲 👍 ⚙️ 🔗 ⏮️ yield 🕐❔ 👫 🥃 ⚙️ 💼.

Info

🔗 ⏮️ yield 🚮 ⏳ FastAPI.

⏮️ ⏬ 👉 🔰 🕴 ✔️ 🖼 ⏮️ 🛠️ & 📤 🎲 📚 🈸 ⚙️ 🛠️ 💽 🎉 🧾.