skills/aaaaqwq/claude-code-skills/backend-development-python

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
GitHub Stars
13
First Seen
Feb 13, 2026
Installed on
codex1