【亲测免费】 FastAPI-Alembic-SQLModel-Async 项目教程

项目介绍

FastAPI-Alembic-SQLModel-Async 是一个项目模板，它使用 FastAPI、Pydantic 2.0、Alembic 和异步 SQLModel 作为 ORM。这个模板展示了如何使用异步 CRUD 操作，并结合了身份验证和基于角色的访问控制。项目由 Allient 开发团队支持，该团队由经验丰富的专业人员组成，专注于 FastAPI 项目和 NLP。

项目快速启动

环境准备

确保你已经安装了以下工具和库：

• Python 3.7+
• Docker
• Docker Compose

克隆项目

git clone https://github.com/jonra1993/fastapi-alembic-sqlmodel-async.git
cd fastapi-alembic-sqlmodel-async

启动项目

使用 Docker Compose 启动项目：

docker-compose up --build

访问应用

项目启动后，你可以通过以下地址访问 API 文档：

http://localhost:8000/docs

应用案例和最佳实践

异步 CRUD 操作

FastAPI-Alembic-SQLModel-Async 提供了一个完整的异步 CRUD 示例，包括创建、读取、更新和删除操作。以下是一个简单的示例：

from fastapi import FastAPI, HTTPException
from sqlmodel import SQLModel, Field, Session, select
from sqlalchemy.ext.asyncio import create_async_engine

app = FastAPI()

class Item(SQLModel, table=True):
    id: int = Field(default=None, primary_key=True)
    name: str
    description: str = None

engine = create_async_engine("sqlite+aiosqlite:///./test.db")

@app.on_event("startup")
async def startup():
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)

@app.post("/items/")
async def create_item(item: Item):
    async with Session(engine) as session:
        session.add(item)
        await session.commit()
        await session.refresh(item)
        return item

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    async with Session(engine) as session:
        item = await session.get(Item, item_id)
        if not item:
            raise HTTPException(status_code=404, detail="Item not found")
        return item

身份验证和访问控制

项目还展示了如何使用 FastAPI 进行身份验证和基于角色的访问控制。以下是一个简单的示例：

from fastapi import FastAPI, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta

app = FastAPI()

SECRET_KEY = "your-secret-key"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

@app.post("/token")
async def login(form_data: OAuth2Password