🔁 🗄 (🔗) 💽¶

👆 💪 ⚙️ encode/databases ⏮️ FastAPI 🔗 💽 ⚙️ async & await.

⚫️ 🔗 ⏮️:

✳
✳
🗄

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

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

Tip

👆 💪 🛠️ 💭 ⚪️➡️ 📄 🔃 🇸🇲 🐜 (🗄 (🔗) 💽), 💖 ⚙️ 🚙 🔢 🎭 🛠️ 💽, 🔬 👆 FastAPI 📟.

👉 📄 🚫 ✔ 📚 💭, 🌓 😑 💃.

🗄 & ⚒ 🆙 `SQLAlchemy`¶

🗄 SQLAlchemy.
✍ metadata 🎚.
✍ 🏓 notes ⚙️ metadata 🎚.

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

Tip

👀 👈 🌐 👉 📟 😁 🇸🇲 🐚.

databases 🚫 🔨 🕳 📥.

🗄 & ⚒ 🆙 `databases`¶

🗄 databases.
✍ DATABASE_URL.
✍ database 🎚.

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

Tip

🚥 👆 🔗 🎏 💽 (✅ ✳), 👆 🔜 💪 🔀 DATABASE_URL.

✍ 🏓¶

👉 💼, 👥 🏗 🏓 🎏 🐍 📁, ✋️ 🏭, 👆 🔜 🎲 💚 ✍ 👫 ⏮️ ⚗, 🛠️ ⏮️ 🛠️, ♒️.

📥, 👉 📄 🔜 🏃 🔗, ▶️️ ⏭ ▶️ 👆 FastAPI 🈸.

✍ engine.
✍ 🌐 🏓 ⚪️➡️ metadata 🎚.

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

✍ 🏷¶

✍ Pydantic 🏷:

🗒 ✍ (NoteIn).
🗒 📨 (Note).

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

🏗 👫 Pydantic 🏷, 🔢 💽 🔜 ✔, 🎻 (🗜), & ✍ (📄).

, 👆 🔜 💪 👀 ⚫️ 🌐 🎓 🛠️ 🩺.

🔗 & 🔌¶

✍ 👆 FastAPI 🈸.
✍ 🎉 🐕‍🦺 🔗 & 🔌 ⚪️➡️ 💽.

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

✍ 🗒¶

✍ ➡ 🛠️ 🔢 ✍ 🗒:

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

Note

👀 👈 👥 🔗 ⏮️ 💽 ⚙️ await, ➡ 🛠️ 🔢 📣 ⏮️ async.

👀 `response_model=List[Note]`¶

⚫️ ⚙️ typing.List.

👈 📄 (& ✔, 🎻, ⛽) 🔢 💽, list NoteⓂ.

✍ 🗒¶

✍ ➡ 🛠️ 🔢 ✍ 🗒:

from typing import List

import databases
import sqlalchemy
from fastapi import FastAPI
from pydantic import BaseModel

# SQLAlchemy specific code, as with any other app
DATABASE_URL = "sqlite:///./test.db"
# DATABASE_URL = "postgresql://user:password@postgresserver/db"

database = databases.Database(DATABASE_URL)

metadata = sqlalchemy.MetaData()

notes = sqlalchemy.Table(
    "notes",
    metadata,
    sqlalchemy.Column("id", sqlalchemy.Integer, primary_key=True),
    sqlalchemy.Column("text", sqlalchemy.String),
    sqlalchemy.Column("completed", sqlalchemy.Boolean),
)


engine = sqlalchemy.create_engine(
    DATABASE_URL, connect_args={"check_same_thread": False}
)
metadata.create_all(engine)


class NoteIn(BaseModel):
    text: str
    completed: bool


class Note(BaseModel):
    id: int
    text: str
    completed: bool


app = FastAPI()


@app.on_event("startup")
async def startup():
    await database.connect()


@app.on_event("shutdown")
async def shutdown():
    await database.disconnect()


@app.get("/notes/", response_model=List[Note])
async def read_notes():
    query = notes.select()
    return await database.fetch_all(query)


@app.post("/notes/", response_model=Note)
async def create_note(note: NoteIn):
    query = notes.insert().values(text=note.text, completed=note.completed)
    last_record_id = await database.execute(query)
    return {**note.dict(), "id": last_record_id}

Note

👀 👈 👥 🔗 ⏮️ 💽 ⚙️ await, ➡ 🛠️ 🔢 📣 ⏮️ async.

🔃 `{**note.dict(), "id": last_record_id}`¶

note Pydantic Note 🎚.

note.dict() 📨 dict ⏮️ 🚮 💽, 🕳 💖:

{
    "text": "Some note",
    "completed": False,
}

✋️ ⚫️ 🚫 ✔️ id 🏑.

👥 ✍ 🆕 dict, 👈 🔌 🔑-💲 👫 ⚪️➡️ note.dict() ⏮️:

{**note.dict()}

**note.dict() "unpacks" the key value pairs directly, so, {**note.dict()} would be, more or less, a copy of note.dict().

& ⤴️, 👥 ↔ 👈 📁 dict, ❎ ➕1️⃣ 🔑-💲 👫: "id": last_record_id:

{**note.dict(), "id": last_record_id}

, 🏁 🏁 📨 🔜 🕳 💖:

{
    "id": 1,
    "text": "Some note",
    "completed": False,
}

✅ ⚫️¶

👆 💪 📁 👉 📟, & 👀 🩺 http://127.0.0.1:8000/docs.

📤 👆 💪 👀 🌐 👆 🛠️ 📄 & 🔗 ⏮️ ⚫️:

🌅 ℹ¶

👆 💪 ✍ 🌅 🔃 encode/databases 🚮 📂 📃.

🔁 🗄 (🔗) 💽¶

🗄 & ⚒ 🆙 SQLAlchemy¶

🗄 & ⚒ 🆙 databases¶

✍ 🏓¶

✍ 🏷¶

🔗 & 🔌¶

✍ 🗒¶

👀 response_model=List[Note]¶

✍ 🗒¶

🔃 {**note.dict(), "id": last_record_id}¶

✅ ⚫️¶

🌅 ℹ¶

🗄 & ⚒ 🆙 `SQLAlchemy`¶

🗄 & ⚒ 🆙 `databases`¶

👀 `response_model=List[Note]`¶

🔃 `{**note.dict(), "id": last_record_id}`¶