python 后端服务

pip install Django
# 检查是否安装成功
django-admin --version

django-admin startproject myproject
cd myproject
# 运行开发服务器
python manage.py runserver

# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('', include('myapp.urls')),  # 包含 myapp 的 URL 配置
]

# api/urls.py
from django.urls import path
from . import views

urlpatterns = [
    path('hello/', views.hello),
]

# api/views.py
from django.http import HttpResponse

def hello(request):
    return HttpResponse("Hello, Django!")

pip install celery redis
# myproject/celery.py
import os
from celery import Celery

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'myproject.settings')
app = Celery('myproject', broker='redis://localhost:6379/0')
app.config_from_object('django.conf:settings', namespace='CELERY')
app.autodiscover_tasks()
# api/tasks.py
from celery import shared_task

@shared_task
def send_email():
    print("Sending email...")

celery -A myproject worker --loglevel=info

pip install mysqlclient  # 或 pymysql
# myproject/settings.py
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.mysql',
        'NAME': 'mydb',
        'USER': 'root',
        'PASSWORD': '123456',
        'HOST': 'localhost',
        'PORT': '3306',
    }
}

# api/utils/db.py
from django.db import connection

def query_data(sql):
    with connection.cursor() as cursor:
        cursor.execute(sql)
        rows = cursor.fetchall()
    return rows

#  使用 ORM 查询数据
from django.db import models

class User(models.Model):
    name = models.CharField(max_length=100)
    email = models.EmailField()

# 查询数据
users = User.objects.all()

pip install fastapi
pip install uvicorn


# 安装 pyproject.toml 核心依赖 
pip install -e .

# 安装核心 + 开发依赖
pip install -e ".[dev]"

# 安装核心 + 测试依赖
pip install -e ".[test]"

pip install poetry
poetry init

poetry add fastapi

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

uvicorn main:app --reload    

project/
│
├── main.py
├── routers/
│   ├── __init__.py
│   └── items.py
└── models/
    ├── __init__.py
    └── items.py

# routers/items.py
from fastapi import APIRouter

# 创建一个 APIRouter 实例
router = APIRouter()

# 定义一个路由
@router.get("/items/")
async def read_items():
  return {"message": "This is a list of items."}

# 定义另一个路由
@router.get("/items/{item_id}")
async def read_item(item_id: int):
  return {"item_id": item_id, "message": f"Item {item_id} is selected."}

# user.py
from fastapi import APIRouter
# 给模块添加前缀和标签
router = APIRouter( 
  prefix="/user",
  tags=["User"]
)
@router.get("/deatails/")
async def user_items():
  return {"message": "This is a deatail of user."}

# routers/__init__.py
from .items import router as items_router  
from .user import router as user_router  

# main.py
from fastapi import FastAPI
from routers import items_router,user_router
"""
引入多个路由模块，然后在下面挂载
"""

# 创建 FastAPI 应用实例
app = FastAPI()

# 挂载路由模块
app.include_router(items_router)

if __name__ == "__main__":
  import uvicorn
  uvicorn.run(app, host="0.0.0.0", port=8000)

from fastapi import FastAPI

app = FastAPI()

@app.get("/items/")
async def read_items():
    return [{"name": "Item 1"}, {"name": "Item 2"}]

@app.post("/items/")
async def create_item(item: Item):  # 请求体为create_item 函数参数
    return item

@app.put("/items/{item_id}")
async def update_item(item_id: int, item: Item):
    return {"item_id": item_id, "item": item}

@app.delete("/items/{item_id}")
async def delete_item(item_id: int):
    return {"item_id": item_id}

from fastapi import FastAPI
from fastapi import Query
app = FastAPI()

# 查询参数
@app.get("/items/")
async def read_items(page: int = 1, limit: int = 10, search: str = None):
    return {
        "page": page,
        "limit": limit,
        "search": search,
    }
# 访问 http://127.0.0.1:8000/items/?page=2&limit=20&search=apple
# 参数默认是可选的（如果未传参，使用默认值）

@app.get("/items/")
async def read_items(page: int = Query(..., gt=0), limit: int = Query(10, le=100)):
# Query(..., gt=0) 表示 page 是必传的，且必须 >0
# Query(10, le=100) 表示 limit 默认 10，且必须 <=100    
    

# 路径参数
@app.get("/items/{item_id}")
async def read_item(item_id: int, q: str = None):
# item_id 是路径参数，直接从 URL 中提取   q 是查询参数（可选）。    
# http://127.0.0.1:8000/items/42?q=test


#  混合使用路径参数 + 查询参数
@app.get("/users/{user_id}/orders")
async def get_user_orders(
    user_id: int,
    page: int = 1,
    limit: int = 10,
    status: str = "pending"
):

# http://127.0.0.1:8000/users/123/orders?page=2&limit=5&status=completed


# 使用 Pydantic 模型接收多个参数（推荐）
from pydantic import BaseModel
from fastapi import Depends

class Pagination(BaseModel):
    page: int = 1
    limit: int = 10

@app.get("/articles/")
async def get_articles(pagination: Pagination = Depends()):
# http://127.0.0.1:8000/articles/?page=3&limit=20


# 接收数组/列表参数
from typing import List
from fastapi import Query

@app.get("/items/")
async def read_items(tags: List[str] = Query(...)):
# http://127.0.0.1:8000/items/?tags=python&tags=fastapi&tags=web    

pip install sqlalchemy  pymysql

from fastapi import FastAPI, Depends
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库连接配置，包含连接池相关参数
SQLALCHEMY_DATABASE_URL = "mysql+mysqlconnector://username:password@localhost:3306/dbname"

# 创建数据库引擎，配置连接池
engine = create_engine(
    SQLALCHEMY_DATABASE_URL,
    pool_size=5,  # 连接池中的初始连接数
    max_overflow=10,  # 连接池允许的最大额外连接数
    pool_timeout=30,  # 从连接池获取连接的超时时间（秒）
    pool_recycle=3600  # 连接在池中保持活动的时间（秒），超时后会被回收
)

# 创建会话工厂
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建基类
Base = declarative_base()


# 定义数据模型
class Item(Base):
    __tablename__ = "items"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(255), index=True)
    description = Column(String(255))


# 创建数据库表
Base.metadata.create_all(bind=engine)

# 创建 FastAPI 应用
app = FastAPI()


# 依赖项，用于获取数据库会话
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


# 简单的 API 端点，用于创建一个新的 Item
@app.post("/items/")
def create_item(name: str, description: str, db: sessionmaker = Depends(get_db)):
    new_item = Item(name=name, description=description)
    db.add(new_item)
    db.commit()
    db.refresh(new_item)
    return new_item