빠르게 변화하는 웹 개발 세계에서는 성능이 무엇보다 중요합니다. 효율적인 캐싱 메커니즘은 중복 계산 및 데이터베이스 쿼리를 줄여 API의 응답성을 크게 향상시킬 수 있습니다. 이 기사에서는 캐싱 및 동시성 제어를 구현하기 위해 SQLModel 및 Redis를 사용하여 py-cachify 라이브러리를 FastAPI 애플리케이션에 통합하는 방법을 살펴보겠습니다.
캐싱은 비용이 많이 드는 작업의 결과를 저장하고 빠른 액세스 스토리지에서 제공하여 웹 애플리케이션의 성능을 향상시키는 강력한 기술입니다. py-cachify를 사용하면 Redis를 스토리지로 활용하여 FastAPI 애플리케이션에 캐싱을 원활하게 추가할 수 있습니다. 또한 py-cachify는 동시성 제어를 위한 도구를 제공하여 중요한 작업 중에 경합 상태를 방지합니다.
이 튜토리얼에서는 ORM용 SQLModel 및 캐싱용 Redis를 사용하여 FastAPI 애플리케이션에서 py-cachify 라이브러리를 설정하는 과정을 안내합니다.
먼저 프로젝트 환경을 설정해 보겠습니다.
시를 통해 새로운 프로젝트 시작:
# 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
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)
수명 내에서 우리는:
데이터베이스와 상호작용하기 위한 간단한 사용자 모델을 생성하겠습니다.
# 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에서 지원하는 모든 데이터베이스를 사용할 수 있습니다.
사용자 모델과 상호작용할 엔드포인트를 만들어 보겠습니다.
# 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 데코레이터 사용:
사용자 데이터가 업데이트되면 클라이언트가 최신 정보를 받을 수 있도록 캐시를 재설정해야 합니다. 이를 실현하기 위해 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)를 호출하여 다음을 수행합니다.
아래에는 캐시된 데코레이터가 .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
한 번으로:
선택적으로 잠금이 이미 획득된 경우 예외를 발생시키거나 특정 값을 반환하도록 @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의 GitHub 저장소를 확인하고 공식 문서에서 더 자세한 안내, 튜토리얼, 예시를 확인하세요.
GitHub 여기에서 이 튜토리얼의 전체 코드에 액세스할 수 있습니다. 자유롭게 저장소를 복제하고 프로젝트 요구 사항에 맞게 구현해 보세요.
py-cachify가 도움이 된다면 GitHub에서 별점을 주어 프로젝트를 지원해 보세요! 귀하의 지원은 추가적인 개선과 새로운 기능을 추진하는 데 도움이 됩니다.
즐거운 코딩하세요!
위 내용은 FastAPI 효율성 극대화: py-cachify를 통한 놀라울 정도로 빠른 캐싱 및 잠금 구현의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!
![[웹 프런트엔드] Node.js 빠른 시작](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
