FastAPI 효율성 극대화: py-cachify를 통한 놀라울 정도로 빠른 캐싱 및 잠금 구현

빠르게 변화하는 웹 개발 세계에서는 성능이 무엇보다 중요합니다. 효율적인 캐싱 메커니즘은 중복 계산 및 데이터베이스 쿼리를 줄여 API의 응답성을 크게 향상시킬 수 있습니다. 이 기사에서는 캐싱 및 동시성 제어를 구현하기 위해 SQLModel 및 Redis를 사용하여 py-cachify 라이브러리를 FastAPI 애플리케이션에 통합하는 방법을 살펴보겠습니다.

목차:

• 소개
• 프로젝트 설정
• SQLModel을 사용하여 데이터베이스 모델 생성
• FastAPI 엔드포인트 구축
• 엔드포인트 결과 캐싱
• 업데이트 엔드포인트 실행 잠금
• 애플리케이션 실행
• 결론

소개

캐싱은 비용이 많이 드는 작업의 결과를 저장하고 빠른 액세스 스토리지에서 제공하여 웹 애플리케이션의 성능을 향상시키는 강력한 기술입니다. py-cachify를 사용하면 Redis를 스토리지로 활용하여 FastAPI 애플리케이션에 캐싱을 원활하게 추가할 수 있습니다. 또한 py-cachify는 동시성 제어를 위한 도구를 제공하여 중요한 작업 중에 경합 상태를 방지합니다.

이 튜토리얼에서는 ORM용 SQLModel 및 캐싱용 Redis를 사용하여 FastAPI 애플리케이션에서 py-cachify 라이브러리를 설정하는 과정을 안내합니다.

프로젝트 설정

먼저 프로젝트 환경을 설정해 보겠습니다.

전제 조건

• 파이썬 3.12
• 시(원하는 패키지 관리자를 사용할 수 있음)
• 로컬에서 실행되거나 원격으로 액세스할 수 있는 Redis 서버

종속성 설치

시를 통해 새로운 프로젝트 시작:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

• FastAPI: API 구축을 위한 웹 프레임워크.
• SQLModel aiosqlite: ORM 및 데이터 유효성 검사를 위해 SQLAlchemy와 Pydantic을 결합합니다.
• Redis: Redis와 상호작용하기 위한 Python 클라이언트
• py-cachify: 캐싱 및 잠금 유틸리티.

py-cachify 초기화 중

py-cachify를 사용하기 전에 Redis 클라이언트로 초기화해야 합니다. FastAPI의 수명 매개변수를 사용하여 이 작업을 수행하겠습니다.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

수명 내에서 우리는:

• 비동기 Redis 클라이언트를 생성합니다.
• 이 클라이언트로 py-cachify를 초기화합니다.

SQLModel을 사용하여 데이터베이스 모델 생성

데이터베이스와 상호작용하기 위한 간단한 사용자 모델을 생성하겠습니다.

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

데이터베이스 엔진을 설정하고 수명 함수에서 테이블을 생성합니다.

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

참고: 단순성을 위해 SQLite를 사용하고 있지만 SQLAlchemy에서 지원하는 모든 데이터베이스를 사용할 수 있습니다.

FastAPI 엔드포인트 구축

사용자 모델과 상호작용할 엔드포인트를 만들어 보겠습니다.

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

엔드포인트 결과 캐싱

이제 불필요한 데이터베이스 쿼리를 피하기 위해 read_user 엔드포인트의 결과를 캐시해 보겠습니다.

엔드포인트 코드는 다음과 같습니다.

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

@cached 데코레이터 사용:

• user_id를 사용하여 고유 키를 지정합니다.
• TTL(Time-To-Live)을 5분(300초)으로 설정합니다.
• 5분 이내에 동일한 user_id를 사용하여 이 엔드포인트를 호출하면 캐시된 결과가 반환됩니다.

업데이트 시 캐시 재설정

사용자 데이터가 업데이트되면 클라이언트가 최신 정보를 받을 수 있도록 캐시를 재설정해야 합니다. 이를 실현하기 위해 update_user 엔드포인트를 수정해 보겠습니다.

# app/db.py
from sqlmodel import Field, SQLModel


class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

read_user.reset(user_id=user_id)를 호출하여 다음을 수행합니다.

• 특정 user_id에 대해 캐시된 데이터를 지웁니다.
• 후속 GET 요청이 데이터베이스에서 최신 데이터를 가져왔는지 확인하세요.

아래에는 캐시된 데코레이터가 .reset 메소드를 추가하여 함수를 동적으로 래핑합니다. 이 방법은 함수의 서명과 유형을 모방하므로 원래 함수에 따라 동기 또는 비동기가 되며 동일한 인수를 허용합니다.

.reset 메소드는 캐시된 데코레이터에 정의된 것과 동일한 키 생성 로직을 사용하여 무효화할 캐시된 항목을 식별합니다. 예를 들어 캐싱 키 패턴이 user-{user_id}인 경우 wait read_user.reset(user_id=123)을 호출하면 user_id=123에 대한 캐시 항목을 구체적으로 대상으로 지정하고 삭제합니다.

업데이트 끝점 실행 잠금

업데이트 중 경쟁 조건을 방지하기 위해 Once 데코레이터를 사용하여 업데이트 엔드포인트의 실행을 잠급니다.

# app/db.py

# Adjust imports
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine


# Add the following at the end of the file
sqlite_file_name = 'database.db'
sqlite_url = f'sqlite+aiosqlite:///{sqlite_file_name}'
engine = create_async_engine(sqlite_url, echo=True)
session_maker = async_sessionmaker(engine)


# app/main.py
# Adjust imports and lifespan function
from sqlmodel import SQLModel

from .db import engine


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        async_client=from_url('redis://localhost:6379/0'),
    )
    # Create SQL Model tables
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

    yield

한 번으로:

• user_id에 따라 기능을 잠급니다.
• 다른 요청이 동일한 사용자를 동시에 업데이트하려고 하면 즉시 226 IM 사용 상태 코드와 함께 응답을 반환합니다.
• 동시 업데이트로 인해 데이터 불일치가 발생하는 것을 방지합니다.

선택적으로 잠금이 이미 획득된 경우 예외를 발생시키거나 특정 값을 반환하도록 @once를 구성할 수 있습니다.

애플리케이션 실행

이제 앱을 실행하고 테스트할 시간입니다!

1) Redis 서버 시작:

Redis 서버가 로컬에서 실행 중인지 또는 원격으로 액세스할 수 있는지 확인하세요. Docker를 사용하여 로컬 Redis 서버를 시작할 수 있습니다.

# app/main.py

# Adjust imports
from fastapi import Depends, FastAPI
from sqlalchemy.ext.asyncio import AsyncSession

from .db import User, engine, session_maker


# Database session dependency
async def get_session():
    async with session_maker() as session:
        yield session


app = FastAPI(lifespan=lifespan)


@app.post('/users/')
async def create_user(user: User, session: AsyncSession = Depends(get_session)) -> User:
    session.add(user)
    await session.commit()
    await session.refresh(user)
    return user


@app.get('/users/{user_id}')
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)


@app.put('/users/{user_id}')
async def update_user(user_id: int, new_user: User, session: AsyncSession = Depends(get_session)) -> User | None:
    user = await session.get(User, user_id)
    if not user:
        return None

    user.name = new_user.name
    user.email = new_user.email

    session.add(user)
    await session.commit()
    await session.refresh(user)

    return user

2) FastAPI 애플리케이션 실행:

모든 설정이 완료되면 Poetry를 사용하여 FastAPI 애플리케이션을 시작할 수 있습니다. 프로젝트의 루트 디렉터리로 이동하여 다음 명령을 실행합니다.

# app/main.py

# Add the import
from py_cachify import cached


@app.get('/users/{user_id}')
@cached('read_user-{user_id}', ttl=300)  # New decorator
async def read_user(user_id: int, session: AsyncSession = Depends(get_session)) -> User | None:
    return await session.get(User, user_id)

3) 캐싱 및 잠금 테스트 및 재생:

캐싱: read_user 함수에 지연(예: asyncio.sleep 사용)을 추가하여 장기 실행 계산을 시뮬레이션합니다. 결과가 캐시되면 응답 시간이 어떻게 대폭 향상되는지 관찰하세요.

예:

# create new project
poetry new --name app py-cachify-fastapi-demo

# enter the directory
cd py-cachify-fastapi-demo

# point poetry to use python3.12
poetry env use python3.12

# add dependencies
poetry add "fastapi[standard]" sqlmodel aiosqlite redis py-cachify

동시성 및 잠금: 마찬가지로, 동시 업데이트 시도가 이루어질 때 잠금 동작을 관찰하기 위해 update_user 함수에 지연을 도입합니다.

예:

# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from py_cachify import init_cachify
from redis.asyncio import from_url


@asynccontextmanager
async def lifespan(_: FastAPI):
    init_cachify(
        # Replace with your redis url if it differs
        async_client=from_url('redis://localhost:6379/0'),
    )
    yield


app = FastAPI(lifespan=lifespan)

이러한 지연은 캐싱으로 인해 후속 읽기가 더 빨라지고 동일한 리소스에 대한 동시 쓰기가 잠금을 통해 효과적으로 관리되어야 하므로 캐싱 및 잠금 메커니즘의 작동 효과를 확인하는 데 도움이 될 수 있습니다.

이제 Postman과 같은 도구를 사용하거나 http://127.0.0.1:8000/docs(앱이 실행 중일 때)로 이동하여 엔드포인트를 테스트하고 성능 개선 및 동시성 제어가 실제로 작동하는 모습을 관찰할 수 있습니다.

향상된 FastAPI 앱을 사용해 보세요!

결론

py-cachify를 FastAPI 애플리케이션에 통합함으로써 API의 성능과 안정성을 모두 향상시키는 수많은 이점을 얻었습니다.

주요 강점을 요약해 보겠습니다.

• 향상된 성능: 반복적인 함수 호출을 캐싱하면 중복 계산과 데이터베이스 적중이 줄어들어 응답 시간이 대폭 향상됩니다.
• 동시성 제어: 내장된 잠금 메커니즘을 통해 py-cachify는 경합 상태를 방지하고 데이터 일관성을 보장합니다. 이는 동시 액세스가 많은 애플리케이션에 매우 중요합니다.
• 유연성: 동기 작업을 하든 비동기 작업을 하든 py-cachify는 원활하게 적응하므로 최신 웹 애플리케이션을 위한 다양한 선택이 가능합니다.
• 사용 편의성: 라이브러리는 FastAPI와 같은 널리 사용되는 Python 프레임워크와 원활하게 통합되므로 최소한의 마찰로 시작할 수 있습니다.
• 전체 유형 주석: py-cachify는 완전한 유형 주석을 제공하므로 최소한의 노력으로 더 우수하고 유지 관리하기 쉬운 코드를 작성할 수 있습니다.
• 최소 설정: 이 튜토리얼에서 설명한 것처럼 py-cachify를 추가하려면 기존 설정 위에 몇 줄만 추가하면 기능을 완전히 활용할 수 있습니다.

더 자세히 알아보고 싶다면 py-cachify의 GitHub 저장소를 확인하고 공식 문서에서 더 자세한 안내, 튜토리얼, 예시를 확인하세요.

GitHub 여기에서 이 튜토리얼의 전체 코드에 액세스할 수 있습니다. 자유롭게 저장소를 복제하고 프로젝트 요구 사항에 맞게 구현해 보세요.

py-cachify가 도움이 된다면 GitHub에서 별점을 주어 프로젝트를 지원해 보세요! 귀하의 지원은 추가적인 개선과 새로운 기능을 추진하는 데 도움이 됩니다.

즐거운 코딩하세요!

위 내용은 FastAPI 효율성 극대화: py-cachify를 통한 놀라울 정도로 빠른 캐싱 및 잠금 구현의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!