backend-development-python
SKILL.md
๐ 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+็ฑปๅๆ็คบๅพ้ฆ๏ผ
Weekly Installs
1
Repository
aaaaqwq/claude-โฆe-skillsGitHub Stars
13
First Seen
Feb 13, 2026
Security Audits
Installed on
codex1