backend-development-python
1
总安装量
1
周安装量
#78292
全站排名
安装命令
npx skills add https://github.com/aaaaqwq/claude-code-skills --skill backend-development-python
Agent 安装分布
codex
1
Skill 文档
ð Python å端å¼åä¸å®¶
èçææç±Pythonäºï¼è¿è¯è¨åèµ·æ¥çtmç½ï¼
ææ¯æ å ¨æ¯
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ
â Python åç«¯ææ¯æ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â Webæ¡æ¶ â
â âââââââââââ âââââââââââ âââââââââââ âââââââââââ â
â â FastAPI â â Django â â Flask â â Tornado â â
â â ç°ä»£ â â ä¼ä¸çº§ â â è½»é级 â â é«å¹¶å â â
â âââââââââââ âââââââââââ âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â æ°æ®å± â
â âââââââââââ âââââââââââ âââââââââââ âââââââââââ â
â âSQLModel â âSQLAlchemyâ â Tortoiseâ â Beanie â â
â â æ°ä¸ä»£ â â ç»å
¸ORM â â 弿¥ORM â â MongoDB â â
â âââââââââââ âââââââââââ âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â æ°æ®éªè¯ â
â âââââââââââ âââââââââââ â
â â Pydanticâ â Msgspec â â
â â ç±»åéªè¯â â è¶
å¿«é â â
â âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â æµè§å¨èªå¨å â
â âââââââââââ âââââââââââ âââââââââââ â
â âPlaywrightâ â Seleniumâ â Scrapy â â
â â ç°ä»£é¦éâ â ç»å
¸ â â ç¬è«æ¡æ¶â â
â âââââââââââ âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â æ°æ®åºè¿ç§» â
â âââââââââââ âââââââââââ â
â â Alembic â â Aerich â â
â â ç»å
¸ â â FastAPI â â
â âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ¤
â è®¤è¯ææ â
â âââââââââââ âââââââââââ âââââââââââ â
â â JWT â â OAuth2 â â FastAPI â â
â â æ ç¶æ â â ç¬¬ä¸æ¹ â â å®å
¨å·¥å
· â â
â âââââââââââ âââââââââââ âââââââââââ â
âââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââââ
FastAPI – ç°ä»£Python APIé¦é
项ç®ç»æï¼æä½³å®è·µï¼
my_project/
âââ app/
â âââ __init__.py
â âââ main.py # åºç¨å
¥å£
â âââ core/
â â âââ __init__.py
â â âââ config.py # é
置管ç
â â âââ security.py # 认è¯ç¸å
³
â â âââ deps.py # ä¾èµæ³¨å
¥
â âââ models/
â â âââ __init__.py
â â âââ user.py # SQLModel模å
â â âââ post.py
â âââ schemas/
â â âââ __init__.py
â â âââ user.py # Pydantic Schema
â â âââ post.py
â âââ api/
â â âââ __init__.py
â â âââ deps.py # APIä¾èµ
â â âââ v1/
â â âââ __init__.py
â â âââ router.py # è·¯ç±èå
â â âââ endpoints/
â â âââ users.py
â â âââ auth.py
â â âââ posts.py
â âââ services/
â â âââ __init__.py
â â âââ user_service.py # ä¸å¡é»è¾
â â âââ auth_service.py
â âââ db/
â â âââ __init__.py
â â âââ session.py # æ°æ®åºä¼è¯
â â âââ init_db.py # åå§å
â âââ utils/
â âââ __init__.py
â âââ logger.py
âââ alembic/ # æ°æ®åºè¿ç§»
â âââ versions/
â âââ env.py
âââ tests/
â âââ __init__.py
â âââ conftest.py
â âââ test_api/
âââ .env.example
âââ pyproject.toml
âââ README.md
æ ¸å¿ï¼Pydantic V2 + SQLModel
# models/user.py
from typing import Optional
from sqlmodel import Field, SQLModel, Relationship
from datetime import datetime
class UserBase(SQLModel):
email: str = Field(index=True, unique=True)
name: str
is_active: bool = True
is_superuser: bool = False
class User(UserBase, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
hashed_password: str
created_at: datetime = Field(default_factory=datetime.utcnow)
updated_at: datetime = Field(default_factory=datetime.utcnow)
posts: list["Post"] = Relationship(back_populates="author")
# schemas/user.py
from pydantic import BaseModel, EmailStr, ConfigDict, Field
class UserBase(BaseModel):
email: EmailStr
name: str = Field(..., min_length=2, max_length=50)
class UserCreate(UserBase):
password: str = Field(..., min_length=8)
class UserUpdate(BaseModel):
name: str | None = Field(None, min_length=2, max_length=50)
email: EmailStr | None = None
class UserResponse(UserBase):
model_config = ConfigDict(from_attributes=True)
id: int
is_active: bool
created_at: datetime
class UserLogin(BaseModel):
email: EmailStr
password: str
宿´çCRUD Service模å¼
# services/user_service.py
from typing import Optional, List
from sqlmodel import Session, select, col
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate
from app.core.security import get_password_hash, verify_password
class UserService:
def __init__(self, session: Session):
self.session = session
async def get(self, user_id: int) -> User | None:
"""è·ååä¸ªç¨æ·"""
return self.session.get(User, user_id)
async def get_by_email(self, email: str) -> User | None:
"""éè¿é®ç®±è·åç¨æ·"""
stmt = select(User).where(User.email == email)
return self.session.exec(stmt).first()
async def get_multi(
self, skip: int = 0, limit: int = 100
) -> List[User]:
"""è·åç¨æ·å表"""
stmt = select(User).offset(skip).limit(limit)
return self.session.exec(stmt).all()
async def create(self, user_in: UserCreate) -> User:
"""åå»ºç¨æ·"""
# æ£æ¥é®ç®±æ¯å¦å·²åå¨
existing = await self.get_by_email(user_in.email)
if existing:
raise ValueError("é®ç®±å·²è¢«æ³¨å")
# åå»ºç¨æ·
user = User.model_validate(
user_in.model_dump(),
update={
"hashed_password": get_password_hash(user_in.password)
}
)
self.session.add(user)
self.session.commit()
self.session.refresh(user)
return user
async def update(
self, user: User, user_in: UserUpdate
) -> User:
"""æ´æ°ç¨æ·"""
update_data = user_in.model_dump(exclude_unset=True)
for field, value in update_data.items():
setattr(user, field, value)
user.updated_at = datetime.utcnow()
self.session.add(user)
self.session.commit()
self.session.refresh(user)
return user
async def delete(self, user: User) -> None:
"""å é¤ç¨æ·"""
self.session.delete(user)
self.session.commit()
async def authenticate(
self, email: str, password: str
) -> User | None:
"""éªè¯ç¨æ·ç»å½"""
user = await self.get_by_email(email)
if not user:
return None
if not verify_password(password, user.hashed_password):
return None
return user
JWTè®¤è¯ + OAuth2
# core/security.py
from datetime import datetime, timedelta
from jose import jwt, JWTError
from passlib.context import CryptContext
from fastapi import HTTPException, status
SECRET_KEY = "your-secret-key-here" # ä»ç¯å¢åé读å
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
def get_password_hash(password: str) -> str:
"""çæå¯ç hash"""
return pwd_context.hash(password)
def verify_password(plain: str, hashed: str) -> bool:
"""éªè¯å¯ç """
return pwd_context.verify(plain, hashed)
def create_access_token(data: dict, expires_delta: timedelta | None = None):
"""å建JWT token"""
to_encode = data.copy()
expire = datetime.utcnow() + (expires_delta or timedelta(minutes=15))
to_encode.update({"exp": expire})
return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
def verify_token(token: str) -> dict | None:
"""éªè¯JWT token"""
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
return payload
except JWTError:
return None
# core/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from sqlmodel import Session
from app.db.session import get_session
from app.models.user import User
from app.core.security import verify_token
from app.services.user_service import UserService
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/api/auth/login")
async def get_current_user(
token: str = Depends(oauth2_scheme),
session: Session = Depends(get_session)
) -> User:
"""è·åå½åç»å½ç¨æ·"""
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="æ æ³éªè¯åè¯",
headers={"WWW-Authenticate": "Bearer"},
)
payload = verify_token(token)
if payload is None:
raise credentials_exception
user_id: int = payload.get("sub")
if user_id is None:
raise credentials_exception
user_service = UserService(session)
user = await user_service.get(user_id)
if user is None:
raise credentials_exception
return user
宿´çAPI端ç¹
# api/v1/endpoints/users.py
from typing import List
from fastapi import APIRouter, Depends, status
from sqlmodel import Session
from app.api.deps import get_current_user
from app.core.deps import get_session
from app.models.user import User
from app.schemas.user import UserResponse, UserCreate, UserUpdate
from app.services.user_service import UserService
router = APIRouter()
@router.get("/", response_model=List[UserResponse])
async def get_users(
skip: int = 0,
limit: int = 100,
session: Session = Depends(get_session),
current_user: User = Depends(get_current_user)
):
"""è·åç¨æ·å表"""
service = UserService(session)
return await service.get_multi(skip, limit)
@router.get("/me", response_model=UserResponse)
async def get_current_user_info(
current_user: User = Depends(get_current_user)
):
"""è·åå½åç¨æ·ä¿¡æ¯"""
return current_user
@router.post("/", response_model=UserResponse, status_code=201)
async def create_user(
user_in: UserCreate,
session: Session = Depends(get_session)
):
"""åå»ºç¨æ·"""
service = UserService(session)
return await service.create(user_in)
@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(
user_id: int,
user_in: UserUpdate,
session: Session = Depends(get_session),
current_user: User = Depends(get_current_user)
):
"""æ´æ°ç¨æ·"""
service = UserService(session)
user = await service.get(user_id)
if not user:
raise HTTPException(status_code=404, detail="ç¨æ·ä¸åå¨")
return await service.update(user, user_in)
Playwright – æµè§å¨èªå¨åé¦é
èçæç¨è¿çætm好ç¨çæµè§å¨èªå¨åå·¥å ·ï¼
å®è£
# å®è£
playwright
pip install playwright
# å®è£
æµè§å¨é©±å¨ï¼å¿
é¡»ï¼ï¼
playwright install
# æè
åªå®è£
ç¹å®æµè§å¨
playwright install chromium
playwright install firefox
playwright install webkit
åºç¡ç¨æ³
from playwright.sync_api import sync_playwright
# 忥API
with sync_playwright() as p:
browser = p.chromium.launch(headless=False) # headless=True为æ 头模å¼
page = browser.new_page()
page.goto("https://example.com")
print(page.title())
browser.close()
# 弿¥APIï¼æ§è½æ´å¥½ï¼æ¨èï¼ï¼
import asyncio
from playwright.async_api import async_playwright
async def main():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
page = await browser.new_page()
await page.goto("https://example.com")
print(await page.title())
await browser.close()
asyncio.run(main())
å¸¸è§æä½
from playwright.async_api import async_playwright
async def browser_operations():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
user_agent="Mozilla/5.0 ..." # èªå®ä¹UA
)
page = await context.new_page()
# 导èª
await page.goto("https://example.com", wait_until="networkidle") # wait_until: load, domcontentloaded, networkidle
await page.go_back()
await page.go_forward()
await page.reload()
# æ¥æ¾å
ç´ ï¼å¤ç§éæ©å¨ï¼
await page.click("button#submit") # CSS
await page.click("text=ç»å½") # ææ¬
await page.click("xpath=//button[@id='submit']") # XPath
await page.click("data-testid=submit") # data屿§
# è¾å
¥æ¡æä½
await page.fill("input[name='email']", "user@example.com")
await page.type("input[name='email']", "user@example.com", delay=100) # 模ææå
await page.clear("input[name='email']")
# 䏿æ¡
await page.select_option("select#country", "China")
# å¤éæ¡/åéæ¡
await page.check("input#agree")
await page.uncheck("input#subscribe")
# ä¸ä¼ æä»¶
await page.set_input_files("input[type='file']", "path/to/file.pdf")
# è·åå
ç´ å±æ§
text = await page.inner_text("div.content")
html = await page.inner_html("div.content")
attr = await page.get_attribute("a#link", "href")
# çå¾
await page.wait_for_selector("div.result", timeout=5000)
await page.wait_for_url("**/success")
await page.wait_for_timeout(1000) # 硬çå¾
ï¼ä¸æ¨èï¼
# æ§è¡JavaScript
result = await page.evaluate("() => document.title")
value = await page.evaluate("el => el.value", await page.query_selector("input"))
# æªå¾
await page.screenshot(path="screenshot.png")
await page.pdf(path="page.pdf") # åªæchromiumæ¯æ
# å¤çå¼¹çª
async with page.expect_popup() as popup_info:
await page.click("a[target='_blank']")
popup = await popup_info.value
await popup.wait_for_load_state()
# å¤çå¯¹è¯æ¡
async with page.expect_dialog() as dialog_info:
await page.click("button")
dialog = await dialog_info.value
await dialog.accept() # æ dialog.dismiss()
await browser.close()
asyncio.run(browser_operations())
ç½é¡µç¬è«å®æ
import asyncio
from playwright.async_api import async_playwright
from typing import List
import json
class Scraper:
def __init__(self):
self.results = []
async def scrape_page(self, url: str) -> dict:
"""ç¬åå个页é¢"""
async with async_playwright() as p:
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# æ¦æªè¯·æ±ï¼åªå è½½å¿
è¦èµæºï¼å éï¼
await page.route("**/*.{png,jpg,jpeg,gif,svg,css,woff,woff2}", lambda route: route.abort())
await page.goto(url, wait_until="domcontentloaded")
# çå¾
æ°æ®å è½½
await page.wait_for_selector(".item")
# æåæ°æ®
items = await page.query_selector_all(".item")
data = []
for item in items:
title = await item.query_selector(".title")
price = await item.query_selector(".price")
data.append({
"title": await title.inner_text() if title else "",
"price": await price.inner_text() if price else "",
})
await browser.close()
return {"url": url, "data": data}
async def scrape_multiple(self, urls: List[str]) -> List[dict]:
"""å¹¶åç¬åå¤ä¸ªé¡µé¢"""
tasks = [self.scrape_page(url) for url in urls]
return await asyncio.gather(*tasks)
# 使ç¨
async def main():
scraper = Scraper()
urls = [
"https://example.com/page/1",
"https://example.com/page/2",
"https://example.com/page/3",
]
results = await scraper.scrape_multiple(urls)
print(json.dumps(results, ensure_ascii=False, indent=2))
asyncio.run(main())
表åèªå¨å¡«å
from playwright.async_api import async_playwright
async def auto_fill_form():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False, slow_mo=100) # slow_mo模æäººç±»æä½é度
page = await browser.new_page()
await page.goto("https://example.com/register")
# å¡«å表å
await page.fill("input#name", "å¼ ä¸")
await page.fill("input#email", "zhangsan@example.com")
await page.fill("input#password", "password123")
await page.fill("input#password-confirm", "password123")
# éæ©æ§å«ï¼åéæ¡ï¼
await page.check("input[value='male']")
# éæ©å
´è¶£ï¼å¤éæ¡ï¼
await page.check("input[value='reading']")
await page.check("input[value='coding']")
# éæ©åå¸ï¼ä¸ææ¡ï¼
await page.select_option("select#city", "Beijing")
# ä¸ä¼ 头å
await page.set_input_files("input#avatar", "avatar.jpg")
# åææ¡æ¬¾
await page.check("input#agree")
# æäº¤åéªè¯
await page.wait_for_selector("button[type='submit']:not([disabled])")
# æäº¤
async with page.expect_response("**/api/register") as response_info:
await page.click("button[type='submit']")
response = await response_info.value
if response.ok:
print("注åæåï¼")
else:
print(f"注å失败ï¼{await response.text()}")
await page.wait_for_timeout(2000) # çå°ç»æ
await browser.close()
asyncio.run(auto_fill_form())
å¤çç»å½åSession
from playwright.async_api import async_playwright
async def login_and_scrape():
async with async_playwright() as p:
browser = await p.chromium.launch(headless=False)
# å建contextï¼ç¸å½äºæµè§å¨é
ç½®æä»¶ï¼å¯ä¿åcookiesï¼
context = await browser.new_context()
page = await context.new_page()
# ç»å½
await page.goto("https://example.com/login")
await page.fill("input#email", "user@example.com")
await page.fill("input#password", "password123")
await page.click("button[type='submit']")
# çå¾
ç»å½æåï¼è·³è½¬å°é¦é¡µæç¹å®å
ç´ åºç°ï¼
await page.wait_for_url("**/dashboard")
# æè
await page.wait_for_selector(".user-avatar")
# ä¿åsessionç¶æï¼ä¸æ¬¡å¯ä»¥ç´æ¥ä½¿ç¨ï¼
await context.storage_state(path="auth.json")
# ä¹åçææè¯·æ±é½å¸¦ç»å½ç¶æ
await page.goto("https://example.com/protected-page")
content = await page.inner_text(".protected-content")
print(content)
# æ¢å¤å·²æsession
# context = await browser.new_context(storage_state="auth.json")
await browser.close()
asyncio.run(login_and_scrape())
åç¬è«å¯¹ç
from playwright.async_api import async_playwright
import random
async def stealth_browser():
async with async_playwright() as p:
browser = await p.chromium.launch(
headless=False,
args=[
'--disable-blink-features=AutomationControlled', # éèèªå¨åç¹å¾
'--disable-dev-shm-usage',
'--no-sandbox',
]
)
context = await browser.new_context(
viewport={"width": 1920, "height": 1080},
locale="zh-CN",
timezone_id="Asia/Shanghai",
# éæºUser-Agent
user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 ..."
)
# 注å
¥èæ¬éèwebdriverç¹å¾
await context.add_init_script("""
Object.defineProperty(navigator, 'webdriver', {
get: () => undefined
});
""")
page = await context.new_page()
# éæºå»¶è¿æ¨¡æäººç±»æä½
async def human_click(selector: str):
await page.wait_for_selector(selector)
await page.wait_for_timeout(random.randint(500, 2000))
await page.click(selector)
# éæºå»¶è¿æ¨¡æäººç±»è¾å
¥
async def human_type(selector: str, text: str):
await page.wait_for_selector(selector)
await page.click(selector)
for char in text:
await page.type(selector, char, delay=random.randint(50, 200))
# 使ç¨
await page.goto("https://example.com")
await human_type("input#search", "Python")
await human_click("button#search-btn")
await browser.close()
asyncio.run(stealth_browser())
æµè¯æ¡æ¶éæ
# tests/test_login.py
import pytest
from playwright.async_api import async_playwright, Page
@pytest.fixture
async def browser_page():
"""æ¯ä¸ªæµè¯ç¬ç«çbrowseråpage"""
async with async_playwright() as p:
browser = await p.chromium.launch()
page = await browser.new_page()
yield page
await browser.close()
@pytest.mark.asyncio
async def test_login_success(browser_page: Page):
"""æµè¯ç»å½æå"""
await browser_page.goto("https://example.com/login")
await browser_page.fill("input#email", "test@example.com")
await browser_page.fill("input#password", "password123")
await browser_page.click("button[type='submit']")
# çå¾
跳转å°dashboard
await browser_page.wait_for_url("**/dashboard")
# éªè¯ç»å½æå
assert await browser_page.inner_text(".user-name") == "Test User"
@pytest.mark.asyncio
async def test_login_failure(browser_page: Page):
"""æµè¯ç»å½å¤±è´¥"""
await browser_page.goto("https://example.com/login")
await browser_page.fill("input#email", "test@example.com")
await browser_page.fill("input#password", "wrongpassword")
await browser_page.click("button[type='submit']")
# éªè¯é误æç¤º
error_msg = await browser_page.inner_text(".error-message")
assert "å¯ç é误" in error_msg or "ç¨æ·ä¸åå¨" in error_msg
Playwrightæä½³å®è·µ
from playwright.async_api import async_playwright, BrowserContext
from typing import AsyncGenerator
# 1. 使ç¨Context Pool管çå¤ä¸ªä¼è¯
class ContextPool:
def __init__(self, max_contexts: int = 5):
self.contexts: list[BrowserContext] = []
self.max_contexts = max_contexts
async def get_context(self, browser) -> BrowserContext:
if self.contexts:
return self.contexts.pop()
if len(self.contexts) < self.max_contexts:
return await browser.new_context()
raise Exception("No available contexts")
async def return_context(self, context: BrowserContext):
if len(self.contexts) < self.max_contexts:
self.contexts.append(context)
else:
await context.close()
# 2. 使ç¨Page Object模å¼
class LoginPage:
def __init__(self, page):
self.page = page
self.email_input = "input#email"
self.password_input = "input#password"
self.submit_button = "button[type='submit']"
async def login(self, email: str, password: str):
await self.page.fill(self.email_input, email)
await self.page.fill(self.password_input, password)
await self.page.click(self.submit_button)
# 3. éè¯æºå¶
from functools import wraps
import asyncio
def retry(max_attempts: int = 3, delay: float = 1.0):
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
for attempt in range(max_attempts):
try:
return await func(*args, **kwargs)
except Exception as e:
if attempt == max_attempts - 1:
raise
await asyncio.sleep(delay * (attempt + 1))
return wrapper
return decorator
@retry(max_attempts=3)
async def fragile_operation(page):
await page.goto("https://flaky-site.com")
return await page.title()
Alembic æ°æ®åºè¿ç§»
# alembic/env.py
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from alembic import context
from sqlmodel import SQLModel
# 导å
¥æææ¨¡å
from app.models.user import User
from app.models.post import Post
# this is the Alembic Config object
config = context.config
# è®¾ç½®æ°æ®åºURL
config.set_main_option("sqlalchemy.url", "postgresql://...")
# è§£ææ¨¡å
target_metadata = SQLModel.metadata
# ...å
¶ä½Alembicé
ç½®
# å建è¿ç§»
# alembic revision --autogenerate -m "åå»ºç¨æ·è¡¨"
# æ§è¡è¿ç§»
# alembic upgrade head
# åæ»è¿ç§»
# alembic downgrade -1
æ§è½ä¼åæå·§
1. 使ç¨å¼æ¥ORM
# Tortoise ORM - 弿¥é«æ§è½
from tortoise import Tortoise, fields
class User(fields.Model):
id = fields.IntField(pk=True)
email = fields.CharField(max_length=255, unique=True)
posts: fields.ReverseRelation["Post"]
# åå§å
await Tortoise.init(
db_url="postgres://...",
modules={"models": ["app.models"]}
)
# æ¥è¯¢
user = await User.get(email="user@example.com")
posts = await user.posts.all()
2. è¿æ¥æ± é ç½®
# db/session.py
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
engine = create_async_engine(
"postgresql+asyncpg://...",
echo=False,
pool_size=20, # è¿æ¥æ± 大å°
max_overflow=0, # æå¤§æº¢åºè¿æ¥æ°
pool_pre_ping=True, # è¿æ¥åæ£æ¥
pool_recycle=3600, # è¿æ¥åæ¶æ¶é´
)
async_session = sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
3. 使ç¨Redisç¼å
import redis.asyncio as redis
from functools import wraps
import json
redis_client = await redis.from_url("redis://localhost")
def cache(ttl: int = 3600, key_prefix: str = ""):
"""ç¼åè£
饰å¨"""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# çæç¼åkey
cache_key = f"{key_prefix}:{args}:{kwargs}"
# å°è¯ä»ç¼åè·å
cached = await redis_client.get(cache_key)
if cached:
return json.loads(cached)
# æ§è¡å彿°
result = await func(*args, **kwargs)
# åå
¥ç¼å
await redis_client.setex(
cache_key, ttl, json.dumps(result, default=str)
)
return result
return wrapper
return decorator
# 使ç¨
@cache(ttl=1800, key_prefix="user")
async def get_user(user_id: int):
return await User.get(id=user_id)
ä¾èµæ¨èï¼pyproject.tomlï¼
[project]
dependencies = [
# Webæ¡æ¶
"fastapi>=0.109.0",
"uvicorn[standard]>=0.27.0",
# æ°æ®éªè¯
"pydantic>=2.6.0",
"pydantic-settings>=2.1.0",
"email-validator>=2.1.0",
# æ°æ®åº
"sqlmodel>=0.0.14",
"sqlalchemy>=2.0.25",
"asyncpg>=0.29.0", # PostgreSQL弿¥é©±å¨
"alembic>=1.13.0",
# 认è¯
"python-jose[cryptography]>=3.3.0",
"passlib[bcrypt]>=1.7.4",
"python-multipart>=0.0.6",
# æµè§å¨èªå¨å
"playwright>=1.40.0",
# å·¥å
·
"redis>=5.0.1",
"httpx>=0.26.0", # 弿¥HTTP客æ·ç«¯
"celery>=5.3.0", # ä»»å¡éå
]
[project.optional-dependencies]
dev = [
"pytest>=7.4.0",
"pytest-asyncio>=0.23.0",
"pytest-playwright>=0.4.0", # Playwrightæµè¯æä»¶
"httpx>=0.26.0", # æµè¯å®¢æ·ç«¯
"black>=24.0.0",
"ruff>=0.1.0",
"mypy>=1.8.0",
]
èç建议ï¼
- æ°é¡¹ç®ç´æ¥ç¨ FastAPI + SQLModel + Pydantic V2
- 大ååå°ç®¡çç¨ Djangoï¼å¼åå¿«ï¼
- é«å¹¶ååºæ¯èè Tornado + 弿¥ORM
- æµè§å¨èªå¨åç´æ¥ä¸ Playwrightï¼å«tmç¨Seleniumäº
- Playwright弿¥APIæ§è½æ´å¥½ï¼ä¼å 使ç¨
- å«tmå¿äºåç±»åæ³¨è§£ï¼Python 3.12+ç±»åæç¤ºå¾é¦ï¼