FastAPI, Pydantic, Psycopgthe holy trinity untuk API web Python

Bahagian 1: Perbincangan

Masukkan FastAPI

Pertama sekali, ambil tajuk dengan secubit garam.

Jika saya bermula dari awal dengan pembangunan API web Python hari ini, saya mungkin akan melihat dengan lebih teliti pada LiteStar, yang nampaknya saya arkitek yang lebih baik dan dengan struktur tadbir urus projek yang lebih baik.

Tetapi kami mempunyai FastAPI dan ia tidak akan ke mana-mana tidak lama lagi. Saya menggunakannya untuk banyak projek peribadi dan profesional dan masih menikmati kesederhanaannya.

Untuk mendapatkan panduan tentang corak reka bentuk FastAPI, jangan lihat lebih jauh daripada halaman ini.

Mendapatkan Data Pangkalan Data

Walaupun FastAPI hebat pada bahagian 'API' yang sebenar, terdapat satu ketidakpastian yang berterusan untuk saya: cara terbaik untuk mengakses pangkalan data, terutamanya jika kita perlu juga mengendalikan jenis data geospatial.

Jom semak pilihan kami.

Nota 1: kami hanya berminat dengan perpustakaan async di sini, kerana FastAPI ialah ASGI.

Nota 2: Saya hanya akan membincangkan penyambungan ke PostgreSQL, walaupun bahagian perbincangan masih berkaitan dengan pangkalan data lain.

Mudah Untuk Kod | Reka Bentuk Kompleks: ORM

Mengendalikan sambungan pangkalan data anda dan menghuraikan data daripada jadual pangkalan data anda ke dalam objek Python.

• SQLAlchemy2: pesaing terbesar dalam dunia Python ORM. Secara peribadi saya sangat tidak suka sintaks, tetapi masing-masing untuk mereka sendiri.

• TortoiseORM: Saya secara peribadi sangat menyukai ORM async yang diinspirasikan oleh Django ini; ia bersih dan bagus untuk digunakan.

• ORM alternatif: terdapat banyak seperti peewee, PonyORM, dll.

The Middle Ground: Pembina Pertanyaan

Tiada sambungan pangkalan data. Hanya keluarkan SQL mentah daripada pertanyaan berasaskan Python dan hantarkannya kepada pemacu pangkalan data.

• SQLAlchemy Core: pembina pertanyaan SQL teras, tanpa pemetaan ke bahagian objek. Terdapat juga ORM tahap yang lebih tinggi yang dibina di atas pangkalan data yang dipanggil ini yang kelihatan sangat bagus. Saya tertanya-tanya sejauh mana projek itu dibangunkan secara aktif.

• PyPika: Saya tidak tahu banyak tentang yang ini.

Reka Bentuk Mudah: Pemacu Pangkalan Data

• asyncpg: ini ialah pemacu pangkalan data async standard emas untuk Postgres, menjadi salah satu yang pertama memasarkan dan paling berprestasi. Walaupun semua pemacu lain menggunakan libpq perpustakaan C untuk antara muka dengan Postgres, MagicStack memilih untuk menulis semula pelaksanaan tersuai mereka sendiri dan juga menyimpang daripada spesifikasi Python DBAPI. Jika prestasi ialah kriteria utama anda di sini, maka asyncpg mungkin merupakan pilihan terbaik.

• psycopg3: psycopg2 jelas merupakan raja kepada dunia pemacu pangkalan data segerak untuk Python/Postgres. psycopg3 (dijenamakan semula kepada hanya psycopg) ialah lelaran pustaka ini yang seterusnya, tidak segerak sepenuhnya. Perpustakaan ini telah menjadi miliknya sejak beberapa tahun kebelakangan ini & saya ingin membincangkannya dengan lebih lanjut. Lihat blog menarik dari penulis ini tentang hari-hari awal psycopg3.

Perhatikan bahawa jelas terdapat perbincangan yang lebih luas dan lebih konseptual di sini mengenai ORM vs pembina pertanyaan vs SQL mentah. Saya tidak akan membincangkannya di sini.

Model Pendua

Pydantic digabungkan dengan FastAPI dan sangat baik untuk pemodelan, pengesahan dan penyarian respons API.

Jika kami memutuskan untuk menggunakan ORM untuk mendapatkan semula data daripada pangkalan data kami, bukankah agak tidak cekap mengekalkan dua set model pangkalan data dalam penyegerakan? (satu untuk ORM, satu lagi untuk Pydantic)?

Bukankah bagus jika kita boleh menggunakan Pydantic untuk memodelkan pangkalan data?

Ini sebenarnya masalah yang cuba diselesaikan oleh pencipta FastAPI dengan SQLModel perpustakaan.

Walaupun ini boleh menjadi penyelesaian yang bagus untuk masalah itu, saya mempunyai beberapa kebimbangan:

• Adakah projek ini mengalami sindrom penyelenggara tunggal seperti FastAPI?

• Ia masih merupakan projek dan konsep yang agak muda, di mana dokumentasi tidaklah hebat.

• Ia secara intrinsiknya terikat dengan Pydantic dan SQLAlchemy, bermakna penghijrahan pergi akan menjadi amat sukar.

• Untuk pertanyaan yang lebih kompleks, turun ke SQLAlchemy di bawah mungkin diperlukan.

Kembali Kepada Asas

Terlalu banyak pilihan! Lumpuh analisis.

Apabila terdapat ketidakpastian, saya akan menggunakan peraturan berikut: pastikan ia mudah.

SQL telah dicipta 50 tahun yang lalu dan masih merupakan kemahiran utama untuk dipelajari oleh mana-mana pembangun. Sintaksnya secara konsisten mudah difahami dan tidak rumit untuk ditulis untuk kebanyakan kes penggunaan (untuk pengguna ORM yang tegar di luar sana, cubalah, anda mungkin terkejut).

Malah, kami juga boleh menggunakan LLM sumber terbuka hari ini untuk menjana (kebanyakannya berfungsi) pertanyaan SQL dan menjimatkan penaipan anda.

Walaupun ORM dan pembina pertanyaan mungkin datang dan pergi, pemacu pangkalan data berkemungkinan lebih konsisten. Perpustakaan psycopg2 asal telah ditulis hampir 20 tahun yang lalu sekarang dan masih digunakan secara aktif dalam pengeluaran di seluruh dunia.

Menggunakan Psycopg dengan Model Pydantic

Seperti yang dibincangkan, walaupun psycopg mungkin tidak berprestasi seperti asyncpg (implikasi dunia sebenar prestasi teori ini boleh dibahaskan), psycopg memfokuskan pada kemudahan penggunaan dan API yang biasa.

Ciri pembunuh bagi saya ialah Row Factories.

Fungsi ini membolehkan anda memetakan data pangkalan data yang dikembalikan kepada mana-mana objek Python, termasuk kelas data lib standard, model daripada perpustakaan attrs yang hebat, dan ya, model Pydantic!

Bagi saya, ini adalah kompromi pendekatan terbaik: fleksibiliti utama SQL mentah, dengan keupayaan keselamatan pengesahan / jenis Pydantic untuk memodelkan pangkalan data. Psycopg juga mengendalikan perkara seperti sanitasi input berubah-ubah untuk mengelakkan suntikan SQL.

Perlu diambil perhatian bahawa asyncpg juga boleh mengendalikan pemetaan kepada model Pydantic, tetapi lebih kepada penyelesaian daripada ciri terbina dalam. Lihat urutan isu ini untuk mendapatkan butiran. Saya juga tidak tahu sama ada pendekatan ini berfungsi dengan baik dengan perpustakaan pemodelan lain.

Seperti yang saya nyatakan di atas, saya biasanya bekerja dengan data geospatial: kawasan yang sering diabaikan oleh ORM dan pembina pertanyaan. Menjatuhkan ke SQL mentah memberi saya keupayaan untuk menghuraikan dan membongkar data geospatial kerana saya memerlukan jenis yang lebih boleh diterima dalam Python tulen. Lihat artikel berkaitan saya tentang topik ini.

Bahagian 2: Contoh Penggunaan

Cipta Jadual Pangkalan Data

Di sini kami mencipta jadual pangkalan data ringkas yang dipanggil pengguna dalam SQL mentah.

Saya juga akan mempertimbangkan untuk mengendalikan penciptaan pangkalan data dan migrasi menggunakan SQL sahaja, tetapi ini adalah topik untuk artikel lain.

init_db.sql

CREATE TYPE public.userrole AS ENUM (
    'READ_ONLY',
    'STANDARD',
    'ADMIN'
);

CREATE TABLE public.users (
    id integer NOT NULL,
    username character varying,
    role public.userrole NOT NULL DEFAULT 'STANDARD',
    profile_img character varying,
    email_address character varying,
    is_email_verified boolean DEFAULT false,
    registered_at timestamp with time zone DEFAULT now()
);

Modelkan Pangkalan Data Anda Dengan Pydantic

Di sini kami mencipta model yang dipanggil DbUser:

db_models.py

from typing import Optional
from enum import Enum
from datetime import datetime
from pydantic import BaseModel
from pydantic.functional_validators import field_validator
from geojson_pydantic import Feature

class UserRole(str, Enum):
    """Types of user, mapped to database enum userrole."""

    READ_ONLY = "READ_ONLY"
    STANDARD = "STANDARD"
    ADMIN = "ADMIN"

class DbUser(BaseModel):
    """Table users."""

    id: int
    username: str
    role: Optional[UserRole] = UserRole.STANDARD
    profile_img: Optional[str] = None
    email_address: Optional[str] = None
    is_email_verified: bool = False
    registered_at: Optional[datetime]
    # This is a geospatial type I will handle in the SQL
    favourite_place: Optional[dict]

    # DB computed fields (handled in the SQL)
    total_users: Optional[int] = None

    # This example isn't very realistic, but you get the idea
    @field_validator("is_email_verified", mode="before")
    @classmethod
    def i_want_my_ints_as_bools(cls, value: int) -> bool:
        """Example of a validator to convert data type."""
        return bool(value)

Di sini kami mendapat jenis keselamatan dan pengesahan Pydantic.

Kami boleh menambah sebarang bentuk pengesahan atau transformasi data pada model ini apabila data diekstrak daripada pangkalan data.

Menyediakan Psycopg Dengan FastAPI

Kami menggunakan psycopg_pool untuk membuat sambungan pangkalan data dikumpul:

db.py

from fastapi import Request
from psycopg import Connection
from psycopg_pool import AsyncConnectionPool

# You should be using environment variables in a settings file here
from app.config import settings


def get_db_connection_pool() -> AsyncConnectionPool:
    """Get the connection pool for psycopg.

    NOTE the pool connection is opened in the FastAPI server startup (lifespan).

    Also note this is also a sync `def`, as it only returns a context manager.
    """
    return AsyncConnectionPool(
        conninfo=settings.DB_URL.unicode_string(), open=False
    )


async def db_conn(request: Request) -> Connection:
    """Get a connection from the psycopg pool.

    Info on connections vs cursors:
    https://www.psycopg.org/psycopg3/docs/advanced/async.html

    Here we are getting a connection from the pool, which will be returned
    after the session ends / endpoint finishes processing.

    In summary:
    - Connection is created on endpoint call.
    - Cursors are used to execute commands throughout endpoint.
      Note it is possible to create multiple cursors from the connection,
      but all will be executed in the same db 'transaction'.
    - Connection is closed on endpoint finish.
    """
    async with request.app.state.db_pool.connection() as conn:
        yield conn

Seterusnya kami membuka kumpulan sambungan dalam acara jangka hayat FastAPI:

main.py

from contextlib import asynccontextmanager
from fastapi import FastAPI

from .db import get_db_connection_pool

@asynccontextmanager
async def lifespan(app: FastAPI):
    """FastAPI startup/shutdown event."""
    # For this demo I use print, but please use logging!
    print("Starting up FastAPI server.")

    # Create a pooled db connection and make available in app state
    # NOTE we can access 'request.app.state.db_pool' in endpoints
    app.state.db_pool = get_db_connection_pool()
    await app.state.db_pool.open()

    yield

    # Shutdown events
    print("Shutting down FastAPI server.")
    # Here we make sure to close the connection pool
    await app.state.db_pool.close()

Kini apabila apl FastAPI anda dimulakan, anda sepatutnya mempunyai kumpulan sambungan terbuka, bersedia untuk mengambil sambungan dari titik akhir dalam.

Kaedah Pembantu Untuk Model Pydantic

Adalah berguna untuk menambah beberapa kaedah pada model Pydantic untuk kefungsian biasa: mendapatkan satu pengguna, semua pengguna, mencipta pengguna, mengemas kini pengguna, memadamkan pengguna.

Tetapi pertama sekali kita harus mencipta beberapa model Pydantic untuk pengesahan input (untuk mencipta pengguna baharu) dan penyiaran output (tindak balas JSON anda melalui API).

skema_pengguna.py

from typing import Annotated
from pydantic import BaseModel, Field
from pydantic.functional_validators import field_validator
from geojson_pydantic import FeatureCollection, Feature, MultiPolygon, Polygon
from .db_models import DbUser

class UserIn(DbUser):
    """User details for insert into DB."""

    # Exclude fields not required for input
    id: Annotated[int, Field(exclude=True)] = None
    favourite_place: Optional[Feature]

    @field_validator("favourite_place", mode="before")
    @classmethod
    def parse_input_geojson(
        cls,
        value: FeatureCollection | Feature | MultiPolygon | Polygon,
    ) -> Optional[Polygon]:
        """Parse any format geojson into a single Polygon."""
        if value is None:
            return None
        # NOTE I don't include this helper function for brevity
        featcol = normalise_to_single_geom_featcol(value)
        return featcol.get("features")[0].get("geometry")

class UserOut(DbUser):
    """User details for insert into DB."""

    # Ensure it's parsed as a Polygon geojson from db object
    favourite_place: Polygon

    # More logic to append computed values

Kemudian kami boleh menentukan kaedah pembantu kami: satu, semua, cipta:

db_models.py

...previous imports
from typing import Self, Optional
from fastapi.exceptions import HTTPException
from psycopg import Connection
from psycopg.rows import class_row

from .user_schemas import UserIn

class DbUser(BaseModel):
    """Table users."""

    ...the fields

    @classmethod
    async def one(cls, db: Connection, user_id: int) -> Self:
        """Get a user by ID.

        NOTE how the favourite_place field is converted in the db to geojson.
        """
        async with db.cursor(row_factory=class_row(cls)) as cur:
            sql = """
                SELECT
                    u.*,
                    ST_AsGeoJSON(favourite_place)::jsonb AS favourite_place,
                    (SELECT COUNT(*) FROM users) AS total_users
                FROM users u
                WHERE
                    u.id = %(user_id)s
                GROUP BY u.id;
            """

            await cur.execute(
                sql,
                {"user_id": user_id},
            )

            db_project = await cur.fetchone()
            if not db_project:
                raise KeyError(f"User ({user_identifier}) not found.")

            return db_project

    @classmethod
    async def all(
        cls, db: Connection, skip: int = 0, limit: int = 100
    ) -> Optional[list[Self]]:
        """Fetch all users."""
        async with db.cursor(row_factory=class_row(cls)) as cur:
            await cur.execute(
                """
                SELECT
                    *,
                    ST_AsGeoJSON(favourite_place)::jsonb
                FROM users
                OFFSET %(offset)s
                LIMIT %(limit)s;
                """,
                {"offset": skip, "limit": limit},
            )
            return await cur.fetchall()

    @classmethod
    async def create(
        cls,
        db: Connection,
        user_in: UserIn,
    ) -> Optional[Self]:
        """Create a new user."""

        # Omit defaults and empty values from the model
        model_dump = user_in.model_dump(exclude_none=True, exclude_default=True)
        columns = ", ".join(model_dump.keys())
        value_placeholders = ", ".join(f"%({key})s" for key in model_dump.keys())

        sql = f"""
            INSERT INTO users
                ({columns})
            VALUES
                ({value_placeholders})
            RETURNING *;
        """


        async with db.cursor(row_factory=class_row(cls)) as cur:
            await cur.execute(sql, model_dump)
            new_user = await cur.fetchone()

            if new_user is None:
                msg = f"Unknown SQL error for data: {model_dump}"
                print(f"Failed user creation: {model_dump}")
                raise HTTPException(status_code=500, detail=msg)

        return new_user

Penggunaan

routes.py

from typing import Annotated
from fastapi import Depends, HTTPException
from psycopg import Connection

from .main import app
from .db import db_conn
from .models import DbUser
from .user_schemas import UserIn, UserOut

@app.post("/", response_model=UserOut)
async def create_user(
    user_info: UserIn,
    db: Annotated[Connection, Depends(db_conn)],
):
    """Create a new user.

    Here the input is parsed and validated by UserIn
    then the output is parsed and validated by UserOut
    returning the user json data.
    """

    new_user = await DbUser.create(db, user_info)
    if not new_user:
        raise HTTPException(
            status_code=422,
            detail="User creation failed.",
        )

    return new_user

    # NOTE within an endpoint we can also use
    # DbUser.one(db, user_id) and DbUser.all(db)

Ini adalah pendekatan yang saya mula gunakan dalam projek yang saya kekalkan, FMTM, alat untuk mengumpul data lapangan untuk komuniti di seluruh dunia.

Lihat pangkalan kod penuh di sini.
Dan ⭐ jika anda mendapati ini berguna!

Itu sahaja buat masa ini! Saya harap ini membantu seseorang di luar sana ?

Atas ialah kandungan terperinci FastAPI, Pydantic, Psycopgthe holy trinity untuk API web Python. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!