FastAPI, Pydantic, PsycopgPython 웹 API의 삼위일체

1부: 토론

FastAPI 입력

우선 제목에 소금 한 꼬집을 넣어주세요.

오늘 Python 웹 API 개발을 처음부터 시작했다면 아마도 더 나은 아키텍처와 더 나은 프로젝트 거버넌스 구조를 갖춘 LiteStar를 더 자세히 살펴볼 것입니다.

하지만 우리에게는 FastAPI가 있는데 곧 아무데도 가지 않을 것입니다. 저는 이 제품을 개인적이고 전문적인 프로젝트에 많이 사용하면서도 여전히 단순함을 즐깁니다.

FastAPI 디자인 패턴에 대한 가이드는 이 페이지를 참조하세요.

데이터베이스 데이터 검색

FastAPI가 실제 'API' 부분에서는 훌륭함에도 불구하고 나에게는 한 가지 지속적인 불확실성이 있었습니다. 특히 지리 공간 데이터 유형도 처리해야 하는 경우 데이터베이스에 가장 잘 액세스하는 방법은 무엇입니까?

옵션을 검토해 보겠습니다.

참고 1: FastAPI는 ASGI이므로 여기서는 async 라이브러리에만 관심이 있습니다.

참고 2: 토론의 일부는 여전히 다른 데이터베이스와 관련되어 있지만 PostgreSQL 연결에 대해서만 설명하겠습니다.

간단한 코드 | 복잡한 설계: ORM

데이터베이스 연결과 데이터베이스 테이블의 데이터를 Python 개체로 구문 분석하는 작업을 처리합니다.

• SQLAlchemy2: Python ORM 세계에서 가장 큰 경쟁자입니다. 개인적으로 구문이 정말 마음에 안 들지만, 각각의 구문이 다릅니다.

• TortoiseORM: 저는 개인적으로 Django에서 영감을 받은 비동기 ORM을 정말 좋아합니다. 깔끔하고 사용하기 좋아요.

• 대체 ORM: peewee, PonyORM 등이 많이 있습니다.

중간 지점: 쿼리 빌더

데이터베이스 연결이 없습니다. Python 기반 쿼리에서 원시 SQL을 출력하고 데이터베이스 드라이버에 전달하기만 하면 됩니다.

• SQLAlchemy Core: 객체 부분에 대한 매핑이 없는 핵심 SQL 쿼리 빌더입니다. 또한 데이터베이스라고 불리는 매우 보기 좋은 더 높은 수준의 ORM이 구축되어 있습니다. 그런데 프로젝트가 얼마나 활발하게 전개되고 있는지 궁금합니다.

• PyPika: 이건 잘 모르겠어요.

심플한 디자인: 데이터베이스 드라이버

• asyncpg: 이는 Postgres용 표준 비동기 데이터베이스 드라이버로, 최초로 시장에 출시되었으며 성능이 가장 뛰어난 것 중 하나입니다. 다른 모든 드라이버는 C 라이브러리 libpq를 사용하여 Postgres와 인터페이스하는 반면 MagicStack은 자체 사용자 정의 구현을 다시 작성하고 Python DBAPI 사양에서 벗어나는 것을 선택했습니다. 여기에서 성능이 주요 기준이라면 asyncpg가 아마도 최선의 선택일 것입니다.

• psycopg3: 음 psycopg2는 확실히 Python/Postgres용 동기 데이터베이스 드라이버 세계의 왕이었습니다. psycopg3(단순히 psycopg로 브랜드 변경)은 이 라이브러리의 다음 완전 비동기 반복 버전입니다. 이 라이브러리는 최근 몇 년 동안 그 자체로 발전했으며 이에 대해 더 자세히 논의하고 싶습니다. psycopg3의 초기 시절에 대한 저자의 흥미로운 블로그를 살펴보세요.

ORM, 쿼리 빌더, 원시 SQL에 관해 더 광범위하고 개념적인 논의가 분명히 있다는 점에 유의하세요. 여기서는 다루지 않겠습니다.

중복된 모델

Pydantic은 FastAPI와 함께 번들로 제공되며 API 응답 모델링, 검증 및 직렬화에 탁월합니다.

ORM을 사용하여 데이터베이스에서 데이터를 검색하기로 결정한 경우 두 세트의 데이터베이스 모델을 동기화하는 것이 약간 비효율적이지 않나요? (하나는 ORM용, 다른 하나는 Pydantic용)?

Pydantic을 사용하여 데이터베이스를 모델링할 수 있다면 좋지 않을까요?

이것이 바로 FastAPI 작성자가 SQLModel 라이브러리를 사용하여 해결하려고 했던 문제입니다.

이것은 문제에 대한 훌륭한 해결책이 될 수 있지만 몇 가지 우려 사항이 있습니다.

• 이 프로젝트도 FastAPI처럼 단일 관리자 증후군에 시달릴까요?

• 아직 초기 단계의 프로젝트이자 개념이므로 문서화가 훌륭하지 않습니다.

• Pydantic 및 SQLAlchemy와 본질적으로 연결되어 있어 마이그레이션이 매우 어려울 수 있습니다.

• 더 복잡한 쿼리의 경우 아래의 SQLAlchemy로 드롭다운해야 할 수도 있습니다.

기본으로 돌아가기

옵션이 너무 많아요! 분석마비.

불확실한 경우 다음 계율을 사용합니다. 단순함을 유지하세요.

SQL은 50년 전에 발명되었으며 여전히 모든 개발자가 배워야 할 핵심 기술입니다. 구문은 대부분의 사용 사례에서 지속적으로 이해하기 쉽고 작성하기가 복잡하지 않습니다(열성적인 ORM 사용자라면 한번 시도해 보십시오. 놀랄 수도 있습니다).

요즘에는 오픈 소스 LLM을 사용하여 (대부분 작동하는) SQL 쿼리를 생성하고 입력 시간을 절약할 수도 있습니다.

ORM과 쿼리 빌더는 왔다 갔다 할 수 있지만 데이터베이스 드라이버는 더 일관적일 가능성이 높습니다. 원본 psycopg2 라이브러리는 거의 20년 전에 작성되었으며 전 세계적으로 여전히 활발하게 사용되고 있습니다.

Pydantic 모델과 함께 Psycopg 사용

논의한 바와 같이 psycopg는 asyncpg만큼 성능이 좋지 않을 수 있지만(이 이론적 성능이 실제로 미치는 영향은 논쟁의 여지가 있음) psycopg는 사용 편의성과 친숙한 API에 중점을 둡니다.

저에게 있어서 가장 중요한 기능은 Row Factory입니다.

이 기능을 사용하면 반환된 데이터베이스 데이터를 표준 lib 데이터 클래스, 훌륭한 속성 라이브러리의 모델, 그리고 Pydantic 모델을 포함한 모든 Python 객체에 매핑할 수 있습니다!

저에게는 이것이 데이터베이스 모델링을 위한 Pydantic의 검증/유형 안전성 기능과 함께 원시 SQL의 궁극적인 유연성이라는 접근 방식의 최선의 절충안입니다. Psycopg는 SQL 주입을 방지하기 위해 변수 입력 위생과 같은 작업도 처리합니다.

asyncpg는 Pydantic 모델에 대한 매핑도 처리할 수 있지만 내장된 기능보다는 해결 방법에 더 가깝다는 점에 유의해야 합니다. 자세한 내용은 이 문제 스레드를 참조하세요. 또한 이 접근 방식이 다른 모델링 라이브러리와 잘 어울리는지도 모르겠습니다.

위에서 언급했듯이 저는 일반적으로 ORM과 쿼리 빌더에서 종종 무시되는 영역인 지리공간 데이터를 사용하여 작업합니다. 원시 SQL로 이동하면 순수 Python에서 더 수용 가능한 유형이 필요하므로 지리 ​​공간 데이터를 구문 분석하고 구문 분석 해제할 수 있는 기능이 제공됩니다. 이 주제에 대한 내 관련 기사를 참조하세요.

2부: 사용 예

데이터베이스 테이블 생성

여기서 원시 SQL로 user라는 간단한 데이터베이스 테이블을 생성합니다.

SQL만을 사용하여 데이터베이스 생성 및 마이그레이션을 처리하는 것도 고려하고 싶지만 이는 다른 기사의 주제입니다.

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()
);

Pydantic으로 데이터베이스 모델링

여기서 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)

여기서 Pydantic의 유형 안전성과 검증을 얻습니다.

데이터베이스에서 데이터를 추출할 때 이 모델에 모든 형태의 검증 또는 데이터 변환을 추가할 수 있습니다.

FastAPI를 사용하여 Psycopg 설정

우리는 psycopg_pool을 사용하여 풀링된 데이터베이스 연결을 생성합니다.

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

다음으로 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()

이제 FastAPI 앱이 시작되면 내부 엔드포인트에서 연결할 준비가 된 개방형 연결 풀이 있어야 합니다.

Pydantic 모델의 도우미 메서드

한 명의 사용자 가져오기, 모든 사용자 가져오기, 사용자 생성, 사용자 업데이트, 사용자 삭제 등 일반적인 기능을 위해 Pydantic 모델에 몇 가지 메소드를 추가하는 것이 유용할 것입니다.

하지만 먼저 입력 검증(새 사용자 생성) 및 출력 직렬화(API를 통한 JSON 응답)를 위한 Pydantic 모델을 생성해야 합니다.

user_schemas.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

그런 다음 도우미 메소드를 정의할 수 있습니다: one, all, create:

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

용법

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)

이것이 제가 유지 관리하는 프로젝트인 전 세계 커뮤니티를 위한 현장 데이터를 수집하는 도구인 FMTM에서 사용하기 시작한 접근 방식입니다.

여기에서 전체 코드베이스를 확인하세요.
그리고 ⭐이 정보가 유용했다면!

지금은 그게 다입니다! 이것이 누군가에게 도움이 되기를 바랍니다.

위 내용은 FastAPI, Pydantic, PsycopgPython 웹 API의 삼위일체의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!