import socket
sock = socket.socket()
sock.bind(("127.0.0.1",9999))
sock.listen(5)
while True:
    conn, addr = sock.accept()
    data = conn.recv(1024)
    print("获取到的数据为：\n", data)
    conn.send(b"HTTP/1.1 200 OK\nserver:liu\n\nhaole")
    conn.close()

import uvicorn
from fastapi import FastAPI
app = FastAPI()

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

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

@app.get("/home")
@app.post("/post")
@app.put("/update")
@app.delete("/delete")

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, Field, field_validator
from datetime import date
from typing import List, Union, Optional

app = FastAPI()


class User(BaseModel):
    name: str
    age: int = Field(default=0, ge=0, le=100)
    birth: Union[None, date] = None
    friends: List[int] = []
    description: Optional[str] = None

    @field_validator('name')
    def name_mustbe_alpha(cls, value):
        assert value.isalpha(), "名字必须为字母"
        return value


@app.post('/user')
def user(user: User):
    print(user.name)
    return user


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

1. 单个文件上传
@app.post('/file')
async def getfile(file: bytes = File()):
    # 适合小文件上传
    print(file)
    print(len(file))
    return {
        "status": "success",
    }

2. 多个文件上传
@app.post('/files')
async def getfiles(files: List[bytes] = File()):
    # 适合小文件上传
    for file in files:
        print(len(file))
    return {"length": len(files)}

import uvicorn
from fastapi import FastAPI, UploadFile

app = FastAPI()


@app.post("/uploadFile")
def upload_file(file: UploadFile):
    with open(file.filename, "wb") as f:
        f.write(file.file.read())
    return {"status": True}


if __name__ == "__main__":
    uvicorn.run("quickstart:app", host="127.0.0.1", port=8000, reload=True)

@app.post("/request")
async def re(request: Request):
    headers = dict(request.headers)
    query = dict(request.query_params)
    body = await request.body()  # 原始请求体
    return {"headers": headers, "query": query, "body": body.decode()}

{
  "headers": {
    "host": "127.0.0.1:8000",
    "connection": "keep-alive",
    "content-length": "0",
    "sec-ch-ua-platform": "\"Windows\"",
    "user-agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/141.0.0.0 Safari/537.36 Edg/141.0.0.0",
    "accept": "application/json",
    "sec-ch-ua": "\"Microsoft Edge\";v=\"141\", \"Not?A_Brand\";v=\"8\", \"Chromium\";v=\"141\"",
    "sec-ch-ua-mobile": "?0",
    "origin": "http://127.0.0.1:8000",
    "sec-fetch-site": "same-origin",
    "sec-fetch-mode": "cors",
    "sec-fetch-dest": "empty",
    "referer": "http://127.0.0.1:8000/docs",
    "accept-encoding": "gzip, deflate, br, zstd",
    "accept-language": "zh-CN,zh;q=0.9,en;q=0.8,en-GB;q=0.7,en-US;q=0.6"
  },
  "query": {},
  "body": ""
}

 pip install tortoise-orm

from tortoise.models import Model
from tortoise import fields


class Student(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32,description="学生姓名")
    pwd = fields.CharField(max_length=32,description="学生密码")
    son = fields.IntField(max_length=32,description="学生学号")

    # 一对多
    cls = fields.ForeignKeyField("models.Cls", related_name="students")

    # 多对多
    curses = fields.ManyToManyField("models.Curse", related_name="students")


class Cls(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32)


class Course(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32)
    teacher = fields.ForeignKeyField("models.Teacher", related_name="courses")


class Teacher(Model):
    id = fields.IntField(pk=True)
    name = fields.CharField(max_length=32)
    tno = fields.IntField(max_length=32)

TORTOISE_ORM = {
    "connections": {
        "default": {
            "engine": "tortoise.backends.mysql",
            "credentials": {
                "user": "root",
                "password": "",
                "host": "127.0.0.1",
                "database": "fastapi_demo",
                "port": 3306,
                "maxsize": 5,
                "minsize": 1,
                "charset": "utf8mb4",
                "echo": True
            }
        }
    },
    "apps": {
        "models":{
            "models": ["models"],
            "default_connection": "default",
        }
    },
    "use_tz": False,
    "timezone": "Asia/Shanghai",
}

import uvicorn
from fastapi import FastAPI
from tortoise.contrib.fastapi import register_tortoise
from orm import TORTOISE_ORM

app = FastAPI()
register_tortoise(
    app=app,
    config=TORTOISE_ORM
)


if __name__ == "__main__":
    uvicorn.run("quickstart:app", host="127.0.0.1", port=8000, reload=True)

pip install aerich

pip install aiomysql

aerich init -t orm.TORTOISE_ORM

aerich init-db 

aerich migrate

aerich upgrade

aerich downgrade

@app.get("/students")
async def get_students():

    # 1）查询所有，返回一个Queryset
    students = await Student.all()
    for student in students:
        print(student.name, student.pwd)

    # 2）过滤查询filter，返回一个Queryset
    stus = await Student.filter(cls_id=3)

    # 3）过滤查询get，返回第一个符合条件的对象
    stu = await Student.get(name="张三")

    # 4）模糊查询
    stusGt20002 = await Student.filter(son__gte=20002)
    stusbetween = await Student.filter(son__range=[20001, 20002, 20003])
    stusin = await Student.filter(son__in=[20001, 20003])

    # 5) values查询
    stuNameAndSon = await Student.all().values("name", "son") #[{"name":"张三","son":20001}, {...}. {...}]

    return {
        "message": "get all student data"
    }

@app.get("/student")
# 查询外联表cls中id为zhang:cls_id的name字段
async def get_student_clsname():
    # 使用await得到的是查询结果，[Student对象, Student对象, ...]
    zhang1 = await Student.get(name="张三")
    # 不使用await得到一个查询链对象描述要查询的信息,<QuerySet对象>
    zhang2 = Student.filter(name="张三")

    # 获取学生对象cls外联表中的信息
    classname1 = await zhang1.cls.values("name")
    # 完善动作链再执行查询操作
    classname2 = await zhang2.values("cls__name")
    return {
        "方式一": classname1,
        "方式二": classname2
    }

@app.get("/student")
async def get_student_cursesname():
    # 使用 prefetch_related 预加载多对多关系
    students = await Student.all().prefetch_related("curses")

    result = []
    for student in students:
        # 现在可以直接访问 curses，因为已经预加载了
        courses = student.curses
        result.append({
            "student_name": student.name,
            "courses": [{"id": course.id, "name": course.name} for course in courses]
        })

    return result

import uvicorn
from fastapi import FastAPI
from tortoise import Tortoise
from tortoise.contrib.fastapi import register_tortoise
from models import Student
from orm import TORTOISE_ORM
# 响应界面所需库
from fastapi.templating import Jinja2Templates
from fastapi import Request

app = FastAPI()
register_tortoise(
    app=app,
    config=TORTOISE_ORM
)

@app.get("/students")
async def get_students(request: Request):
    templates = Jinja2Templates(directory="templates")
    students = await Student.all()
    return templates.TemplateResponse(
        "index.html", {
            "request": request,
            "students": students
        }
    )

if __name__ == "__main__":
    uvicorn.run("quickstart:app", host="127.0.0.1", port=8000, reload=True)

from typing import List
import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel, field_validator
from tortoise.contrib.fastapi import register_tortoise
from models import Student, Course
from orm import TORTOISE_ORM

app = FastAPI()
register_tortoise(
    app=app,
    config=TORTOISE_ORM
)


class StudentIn(BaseModel):
    name: str
    pwd: str
    son: int
    cls: int
    curses: List[int] = [1, 2]

    @field_validator("name")
    def check_name(cls, v):
        assert v.isalpha(), "name must be alphanumeric"
        return v


@app.post("/student")
async def add(stu: StudentIn):
    # 插入方式一
    newstudent = Student(name=stu.name, pwd=stu.pwd, son=stu.son, cls_id=stu.cls)
    await newstudent.save()

    # 插入方式二
    newstudent = await Student.create(name=stu.name, pwd=stu.pwd, son=stu.son, cls_id=stu.cls)

    # 多对多形式插入
    student_courses = await Course.filter(id__in=stu.curses)
    await newstudent.curses.add(*student_courses)
    return newstudent

@app.put("/student/{stu_id}")
async def update_student(stu_id: int, stu: StudentIn):
    # 将传进的数据转化为json格式
    new_date = stu.model_dump()

    # 将传进来的数据中属于多对多关联表中的curses_id部分弹出给curses_date，后面单独添加
    curses_date = new_date.pop("curses")

    # 使用filter中自带的update方法更新数据，需要解构
    await Student.filter(id=stu_id).update(**new_date)

    '''
    下面是多对多数据添加部分
    1. 获取要编辑的学生对象
    2. 清除该对象中原先关联的curses数据
    3. 将之前获取到的curses_date转化为curses对象集合
    4. 将curses对象集合通过get的add方法添加进关联表
    '''
    # 获取要编辑的学生
    edit_stu =await Student.get(id=stu_id)

    # 清除旧数据
    await edit_stu.curses.clear()
    # 将传递进来的json实例化课程对象
    choose_cursor =await Course.filter(id__in=curses_date)
    # 将课程对象添加到编辑的学生对象的多对多关联表中
    await edit_stu.curses.add(*choose_cursor)

    return stu_id

@app.delete("/student/{stu_id}")
async def delete_student(stu_id: int):
    await Student.filter(id=stu_id).delete()