#FastAPI

import asyncio


async def method01():
    print("协程方法method01开始执行")
    # 模拟一个耗时操作,让当前协程挂起,此时其他协程可以执行
    await asyncio.sleep(2)
    print("协程方法method01结束")


async def method02():
    print("协程方法method02开始执行")
    # 模拟一个耗时操作,让当前协程挂起,此时其他协程可以执行
    await asyncio.sleep(2)
    print("协程方法method02结束")


#定义一个主协程,做统一调度
async def main():
    # 创建并发任务
    task01 = asyncio.create_task(method01())
    task02 = asyncio.create_task(method02())

    # 等待所有任务完成,在执行主协程下面的代码
    await task01
    await task02


if __name__ == '__main__':
   # 调用main函数,启动协程
   asyncio.run(main())

from fastapi import FastAPI
#创建FastAPI对象
app = FastAPI()

@app.get("/")
async def request_method01():
    return {"message":"hello fastapi"}


@app.get("/items/{item_id}")
async def request_method02(item_id:int,param:str = None):
    return {"item_id":item_id,"param":param}

import uvicorn
from fastapi import FastAPI
#创建FastAPI对象
app = FastAPI()

@app.get("/")
async def request_method01():
    return {"message2222":"hello fastapi"}

if __name__ == '__main__':
    uvicorn.run(
        app="demo02_fastapi:app",
        host="0.0.0.0",
        port=8000,
        reload=True #代码修改后自动重启
    )

from fastapi import FastAPI
app = FastAPI()

# 查询
@app.get("/users/{user_id}") 
def get_user(user_id: int):
    return {"id": user_id, "name": "tom"}

@app.post("/users")               # 添加
def create_user(username: str, password: str):
    return {"msg": "添加成功", "username": username}

@app.put("/users/{user_id}")      # 修改
def update_user(user_id: int, username: str):
    return {"msg": "修改成功", "id": user_id}

@app.delete("/users/{user_id}")   # 删除
def delete_user(user_id: int):
    return {"msg": "删除成功", "id": user_id}

import uvicorn
from fastapi import FastAPI

app = FastAPI()

@app.get("/items/main")
async def request_method02():
    return {"main":"main"}


@app.get("/items/{item_id}")
async def request_method01(item_id):
    return {"item_id":item_id}

if __name__ == '__main__':
    uvicorn.run(
        app="demo03_fastapi:app",
        host="0.0.0.0",
        port=8000,
        reload=True #代码修改后自动重启
    )

import uvicorn
from fastapi import FastAPI

app = FastAPI()
@app.get("/items")
async def request_method01(param1:str = None,param2:str = None):
    return {"param1":param1,"param2":param2}


if __name__ == '__main__':
    uvicorn.run(
        app="demo04_fastapi:app",
        host="0.0.0.0",
        port=8000,
        reload=True #代码修改后自动重启
    )

1.方式1:默认值: -> 直接在方法参数上给参数赋值
2.使用:访问路径上不用专门写请求参数
============================================================================
@app.get("/items/")
async def request_method01(param1: str = "tom",param2: str = "jack"):
  return {"param1": param1, "param2": param2}

1.方式2: -> 直接在方法参数上给参数赋值为None
2.使用:
  a.请求路径上不写请求参数,请求参数就是null
  b.请求路径上写请求参数,请求参数就是具体我们赋的值
============================================================================
@app.get("/items/")
async def request_method01(param1: str = None,param2: str = None):
    return {"param1": param1, "param2": param2}

1.方式3:  -> 不要给任何默认值,None也不要给
2.使用:  ?param1=tom&param2=jack
  请求路径上必须要携带请求参数
============================================================================
@app.get("/items/")
async def request_method01(param1: str,param2: str):
    return {"param1": param1, "param2": param2}

加上以下代码:
==============================================
# 定义数据模型类,需要继承 BaseModel 的类。
class Item(BaseModel):
  param1: str = None
  param2: str = None

import uvicorn
from fastapi import FastAPI
from pydantic import BaseModel


# 定义数据模型类,需要继承 BaseModel 的类。
class Item(BaseModel):
  param1: str = None
  param2: str = None

app = FastAPI()

@app.post("/items/")
async def request_method01(item: Item):
  return item

if __name__ == "__main__":
  # 直接在代码中启动uvicorn服务器
  uvicorn.run(
    app="demo05_fastapi:app",    # 指定要运行的FastAPI应用实例
    host="0.0.0.0", # 允许外部访问(本地可通过127.0.0.1或localhost访问)
    port=8000,   # 端口号
    reload=True  # 开发模式：代码修改后自动重启(生产环境需去掉)
)

from fastapi import APIRouter
#创建APIRouter对象,指定路由前缀(/users)
router = APIRouter(
    prefix="/users", #定义用户模块的路由前缀
    tags=["用户管理模块"]
)

@router.get("/")
async def find_all_user():
    return {"message":"查询所有用户"}


@router.get("/{user_id}")
async def find_user_by_id(user_id:int):
    return {"message":f"查询用户id为{user_id}"}

from fastapi import APIRouter
#创建APIRouter对象,指定路由前缀(/users)
router = APIRouter(
    prefix="/items", #定义商品模块的路由前缀
    tags=["商品管理模块"]
)

@router.get("/")
async def find_all_items():
    return {"message":"查询所有商品"}


@router.get("/{item_id}")
async def find_items_by_id(item_id:int):
    return {"message":f"查询商品id为{item_id}"}

from fastapi import FastAPI
from routers import item, user
import uvicorn

app = FastAPI(title="路由分发")

# 挂载用户模块的路由
app.include_router(user.router)

# 挂载商品模块的路由
app.include_router(item.router)


# 给主程序定义一个路由
@app.get("/")
async def index():
    return {"message": "欢迎访问主页面"}


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

from sqlalchemy.ext.declarative import declarative_base

# 生成基类,所有模型需继承该类
Base = declarative_base()

from sqlalchemy import Column, Integer, String, ForeignKey, Date
from sqlalchemy.orm import relationship
from base import Base

"""
部门模型(一对多：一个部门包含多个员工)
"""
class Department(Base):
  # 定义表,声明表名
  __tablename__ = "departments"

  #定义表中的列名以及数据类型以及约束
  id = Column(Integer, primary_key=True, autoincrement=True)
  name = Column(String(50), nullable=False, unique=True) # 部门名称
  location = Column(String(100)) # 部门位置(如"北京总部")

  # 一对多关联员工：返回员工列表,显式指定外键(可选,但更清晰)
  employees = relationship(
    "Employee",
    back_populates="department",
    # 可选：级联删除(删除部门时自动删除下属员工,根据业务选择)
    # cascade="all, delete-orphan",
    # 可选：懒加载策略,优化查询性能
    lazy="selectin"
  )

""" 
员工模型(多对一：多个员工属于一个部门)
"""
class Employee(Base):
  # 定义表,声明表名
  __tablename__ = "employees"

  # 定义表中的列名以及数据类型以及约束
  id = Column(Integer, primary_key=True, autoincrement=True)
  name = Column(String(50), nullable=False) # 姓名
  age = Column(Integer) # 年龄
  hire_date = Column(Date) # 入职日期

  #添加外键约束,绑定到departments.id
  department_id = Column(
    Integer,
    ForeignKey("departments.id", ondelete="RESTRICT"), # 禁止删除有员工的部门
    nullable=False # 员工必须归属一个部门(根据业务可改为True,允许无部门)
  )

  # 核心修正：多对一关联,指定uselist=False(返回单个部门对象)
  department = relationship(
    "Department",
    back_populates="employees",
    lazy="joined" # 可选：立即加载部门数据,减少查询次数
  )

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

# MySQL 连接格式：mysql+pymysql://用户名:密码@主机地址:端口/数据库名
DATABASE_URL = "mysql+pymysql://root:root@localhost:3306/fastapi_db"

# 创建引擎(echo=True 会打印执行的 SQL,方便调试),相当于创建Connection连接
engine = create_engine(
  DATABASE_URL,
  echo=True, # 是否启用日志输出 开发环境启用,生产环境关闭
  pool_pre_ping=True # 连接前检查有效性,避免连接失效
)

# 创建会话工厂(绑定引擎),相当于创建的cursor对象,用于执行sql
SessionLocal = sessionmaker(
  autocommit=False, # 关闭自动提交,需手动 commit()
  bind=engine
)

from base import Base
from database import engine
from models import Employee,Department #必须要加

# 创建表
def create_table():
  print("注册的表名:", Base.metadata.tables.keys())
  # 创建所有模型对应的表
  Base.metadata.create_all(bind=engine)
  print("表创建成功")

if __name__ == '__main__':
  create_table()

#添加功能
def insert_data():
    # 获取数据库会话
    db = SessionLocal()

    try:
        # =========== 第一步：新增部门 =============
        new_dept = Department(
            name="研发部",  # 部门名称(唯一)
            location="北京总部"  # 部门位置
        )
        db.add(new_dept)  # 将部门对象加入会话
        db.commit()  # 提交到数据库(执行 INSERT 语句)
        db.refresh(new_dept)  # 刷新对象,获取自增的 id 等字段

        # =========== 第二步：新增关联的员工 ===========
        # 员工1：关联上面创建的研发部
        emp1 = Employee(
            name="张三",
            age=30,
            hire_date=date(2023, 1, 1),  # 入职日期(datetime.date 类型)
            department_id=new_dept.id  # 关联部门 ID(外键)
        )
        # 员工2：同部门的另一个员工
        emp2 = Employee(
            name="李四",
            age=28,
            hire_date=date(2023, 3, 15),
            department_id=new_dept.id
        )

        # 批量添加员工(也可逐个 add)
        db.add_all([emp1, emp2])
        db.commit()  # 提交员工数据
        # 刷新员工对象,获取自增 ID
        db.refresh(emp1)
        db.refresh(emp2)

        # ===================== 输出结果 =====================
        print(f"新增部门：ID={new_dept.id},名称={new_dept.name},位置={new_dept.location}")
        print(f"新增员工1：ID={emp1.id},姓名={emp1.name},所属部门={new_dept.name}")
        print(f"新增员工2：ID={emp2.id},姓名={emp2.name},所属部门={new_dept.name}")

        # 验证关联关系(通过 ORM 关联查询)
        print("\n【验证关联关系】")
        # 从员工查部门
        print(f"员工{emp1.name}的部门名称：{emp1.department.name}")
        # 从部门查员工
        dept_employees = new_dept.employees
        print(f"部门{new_dept.name}的员工列表：{[emp.name for emp in dept_employees]}")

    except Exception as e:
        db.rollback()  # 出错时回滚
        print(f"新增失败：{e}")
    finally:
        db.close()  # 关闭会话

def delete_data():
    # 获取会话
    session = SessionLocal()

    try:
        # 删除单个员工
        emp = session.query(Employee).filter(Employee.name == "李四").first()
        if emp:
            session.delete(emp)
            session.commit()
            print(f"已删除员工：{emp.name}")

    except Exception as e:
        session.rollback()  # 出错时回滚
        print(f"刪除失败：{e}")
    finally:
        session.close()  # 关闭会话

def update_data():
    # 获取会话
    session = SessionLocal()
    try:
        # 修改员工年龄
        emp = session.query(Employee).filter(Employee.name == "张三").first()
        if emp:
            emp.age = 31  # 直接修改属性
            session.commit()  # 提交更新
            print(f"修改后 {emp.name} 的年龄：{emp.age}")

    except Exception as e:
        session.rollback()  # 出错时回滚
        print(f"修改失败：{e}")
    finally:
        session.close()  # 关闭会话

def find_data():
    # 获取会话
    session = SessionLocal()
    try:
        # ======按主键查询 get========
        #  查询id=1的部门
        dept = session.get(Department, 1)
        print(f"部门 ID=1：{dept.name}({dept.location})")

        # ======过滤(filter)查询========
        # 查询研发部的所有员工
        rd_employees = session.query(Employee).filter(
            Employee.department_id == 1  # 按部门ID过滤
        ).all()
        print("研发部员工：", [emp.name for emp in rd_employees])  # 输出：['张三']

        # 查询年龄>30的员工
        old_employees = session.query(Employee).filter(Employee.age > 30).all()
        print("年龄>30的员工：", [emp.name for emp in old_employees])  # 输出：['张三', '王五']

        # ======逻辑运算(and_/or_)查询========
        from sqlalchemy import and_, or_

        # 年龄30-40且属于研发部的员工(and_)
        emp = session.query(Employee).filter(
            and_(Employee.age.between(30, 40), Employee.department_id == 1)
        ).first()
        print("符合条件的员工：", emp.name)  # 输出：张三

        # 属于市场部或年龄>32的员工(or_)
        emps = session.query(Employee).filter(
            or_(Employee.department_id == 2, Employee.age > 32)
        ).all()
        print("符合条件的员工：", [emp.name for emp in emps])  # 输出：['张三', '王五']

        # ======表连接(join)查询========
        # 内连接：查询员工及其所属部门名称
        # 语法：query(主表, 关联表).join(关联表, 连接条件)
        result = session.query(Employee, Department).join(
            Department, Employee.department_id == Department.id
        ).all()

        for emp, dept in result:
            print(f"员工 {emp.name} 属于 {dept.name}")
        # 输出：
        # 员工 张三 属于 研发部
        # 员工 王五 属于 市场部

        # ======预加载关联数据(joinedload)查询========
        from sqlalchemy.orm import joinedload

        # 加载员工时同时加载部门信息(避免多次查询)
        employees = session.query(Employee).options(
            joinedload(Employee.department)  # 预加载关联的 department
        ).all()

        # 直接访问关联数据,不会触发新查询
        for emp in employees:
            print(f"{emp.name} 的部门：{emp.department.name}")
        # 输出：
        # 张三 的部门：研发部
        # 王五 的部门：市场部

        # ======子查询(subquery)========
        from sqlalchemy import func
        # 子查询：统计每个部门的员工数,再查询员工数>0的部门
        # 步骤1：创建子查询(统计部门员工数)
        dept_emp_count = session.query(
            Employee.department_id,
            func.count(Employee.id).label("count")  # 别名 count
        ).group_by(Employee.department_id).subquery()  # 转为子查询

        # 步骤2：主查询(关联子查询结果)
        depts = session.query(Department).join(
            dept_emp_count, Department.id == dept_emp_count.c.department_id
        ).filter(dept_emp_count.c.count > 0).all()  # 筛选员工数>0的部门

        print("有员工的部门：", [dept.name for dept in depts])  # 输出：['研发部', '市场部']

        # ======去重(distinct)========
        # 查询所有有员工的部门位置(去重)
        locations = session.query(Department.location).join(
            Employee
        ).distinct().all()  # distinct() 去重

        print("部门位置：", [loc[0] for loc in locations])  # 输出：['北京', '广州']

        # ======结果获取(first/all)========
        # first()：返回第一条结果(适合唯一查询)
        first_emp = session.query(Employee).first()
        print("第一个员工：", first_emp.name)  # 输出：张三

        # all()：返回所有结果(列表)
        all_depts = session.query(Department).all()
        print("所有部门：", [dept.name for dept in all_depts])  # 输出：['研发部', '市场部']

    except Exception as e:
        session.rollback()  # 出错时回滚
        print(f"查询失败：{e}")
    finally:
        session.close()  # 关闭会话

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase

class Base(DeclarativeBase):
  pass

class User(Base):
  __tablename__ = "users"
  id = Column(Integer, primary_key=True)
  username = Column(String(50))

  # 一对多：用户拥有多个商品(通过 back_populates 与 Item.owner 双向关联)
  items = relationship(
    "Item", # 关联目标模型
    back_populates="owner", # 反向关联字段(Item 中的 owner)
    lazy="selectin", # 加载方式：查询用户时同时加载商品
    cascade="save-update" # 级联保存：保存用户时自动保存关联商品
  )

class Item(Base):
  __tablename__ = "items"
  id = Column(Integer, primary_key=True)
  title = Column(String(100))
  owner_id = Column(Integer, ForeignKey("users.id")) # 外键关联用户表

  # 多对一：商品属于一个用户(反向关联)
  owner = relationship(
    "User", # 关联目标模型
    back_populates="items" # 反向关联字段(User 中的 items)
  )

# 获取会话
session = SessionLocal()
# 创建用户和商品
user = User(username="test")
item = Item(title="商品1", owner=user) # 直接关联用户
user.items.append(item) # 或通过用户的 items 列表添加

session.add(user)
session.commit()

# 查询关联数据
user = session.query(User).first()
print(user.items) # 获取用户的所有商品(因 lazy="selectin" 已加载)

item = session.query(Item).first()
print(item.owner.username) # 获取商品所属用户的用户名

class Base(DeclarativeBase):
  pass

class User(Base):
  __tablename__ = "users"
  id = Column(Integer, primary_key=True)
  username = Column(String(50))

  # 一对一：用户对应一个资料(uselist=False 表示非集合)
  profile = relationship(
    "Profile",
    back_populates="user",
    uselist=False, # 关键：关联结果为单个对象(非列表)
    cascade="all, delete-orphan" # 删除用户时删除资料
  )

class Profile(Base):
  __tablename__ = "profiles"
  id = Column(Integer, primary_key=True)
  bio = Column(String(200)) # 个人简介
  user_id = Column(Integer, ForeignKey("users.id"), unique=True) # 外键唯一

  # 反向关联用户
  user = relationship("User", back_populates="profile")

session = SessionLocal()
# 方式1：先创建用户，再创建资料并关联
user1 = User(username="alice")
session.add(user1)
session.commit() # 先提交用户，获取 ID

profile1 = Profile(bio="喜欢读书", user_id=user1.id) # 通过 user_id 关联
session.add(profile1)
session.commit()

# 方式2：直接通过 relationship 关联(更简洁)
user2 = User(
  username="bob",
  profile=Profile(bio="热爱运动") # 直接嵌套 Profile 对象
)
session.add(user2)
session.commit() # 自动同步 user_id

from sqlalchemy import Column, Integer, String, ForeignKey
from sqlalchemy.orm import relationship, DeclarativeBase

class Base(DeclarativeBase):
  pass

from sqlalchemy import Table # 用于定义中间表

# 1. 定义中间表(无需模型类，直接用 Table 定义)
student_course = Table(
  "student_course", # 中间表名
  Base.metadata,
  Column("student_id", Integer, ForeignKey("students.id"), primary_key=True),
  Column("course_id", Integer, ForeignKey("courses.id"), primary_key=True)
)

class Student(Base):
  __tablename__ = "students"
  id = Column(Integer, primary_key=True)
  name = Column(String(50))

  # 多对多：学生选修多个课程(通过 secondary 指定中间表)
  courses = relationship(
    "Course",
    secondary=student_course, # 关联中间表
    back_populates="students",
    lazy="selectin"
)

class Course(Base):
  __tablename__ = "courses"
  id = Column(Integer, primary_key=True)
  name = Column(String(100))

  # 反向关联：课程包含多个学生
  students = relationship(
    "Student",
    secondary=student_course,
    back_populates="courses"
  )

# 获取会话
session = SessionLocal()
# 创建学生和课程
student1 = Student(name="张三")
student2 = Student(name="李四")
course1 = Course(name="数学")
course2 = Course(name="英语")

# 建立关联
student1.courses = [course1, course2]
student2.courses = [course1]

session.add_all([student1, student2, course1, course2])
session.commit()

# 查询：学生选修的课程
print(student1.courses) # [Course(name="数学"), Course(name="英语")]

# 查询：课程包含的学生
print(course1.students) # [Student(name="张三"), Student(name="李四")]

import subprocess
import sys

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session

from table_2_models import Departments, Employees

# 创建数据库引擎
db_host = "localhost"
db_port = 3306
db_name = "fastapi_db"
db_user_name = "root"
db_password = "root"
url = f"mysql+pymysql://{db_user_name}:{db_password}@{db_host}:{db_port}/{db_name}?charset=utf8mb4"
engine = create_engine(url, echo=True)

# 配置会话工厂
engine = create_engine(url)
SessionLocal = sessionmaker(bind=engine, autocommit=False, autoflush=False)

# 生成模型类
def table_2_model(run=False):
  """将数据库表映射为Python类"""
  if not run:
    return
  output_path = "table_2_models.py"

  venv_python = sys.executable # 若PyCharm使用虚拟环境，这里会返回.venv下的python.exe
  print("当前使用的Python路径：", venv_python) # 确认输出是.venv/Scripts/python.exe

  cmd = [venv_python, "-m", "sqlacodegen", url]
  result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8")

  # 打印执行结果（定位问题核心）
  print("=== 命令执行结果 ===")
  print(f"返回码（0=成功，非0=失败）：{result.returncode}")
  print(f"标准输出：\n{result.stdout}")
  print(f"错误输出：\n{result.stderr}") # 重点看这里，会显示失败原因

  with open(output_path, "w", encoding="utf-8") as f:
    f.write(result.stdout)

# 向员工和部门表中插入数据
def insert_dept_emp():
  # 1. 创建员工对象（用关键字参数，日期转换为date类型）
  emp = Employees(
    id=100, # 显式指定参数名
    name='zs',
    age=20,
    # hire_date=date(2025, 10, 10) # 字符串转date对象
    hire_date='2025-10-10' # 字符串转date对象
  )

  # 2. 创建部门对象（关键字参数，日期转换为datetime类型）
  dept = Departments(
    id=10,
    name='研发部',
    location='北京',
    created_at='2025-10-10', # 字符串转datetime对象
    employees=[emp] # 关联员工
  )

  # 3. 插入数据库
  with Session(engine) as session:
    session.add(dept)
    try:
      session.commit()
      print(f"插入成功！部门ID：{dept.id}，员工ID：{emp.id}")
    except Exception as e:
      session.rollback()
      print(f"插入失败：{e}")

if __name__ == "__main__":
  table_2_model(True)
  # insert_dept_emp()

import uvicorn
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from datetime import date

from table_2_models import Departments, Employees
from database import SessionLocal, engine

app = FastAPI()

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

# 部门相关接口
@app.post("/departments/")
def create_department(name: str, location: str, db: Session = Depends(get_db)):
  db_department = Departments(name=name, location=location)
  db.add(db_department)
  db.commit()
  db.refresh(db_department)
  return db_department

@app.get("/departments/")
def read_departments(db: Session = Depends(get_db)):
  return db.query(Departments).all()

@app.get("/departments/{department_id}")
def read_department(department_id: int, db: Session = Depends(get_db)):
  department = db.query(Departments).filter(Departments.id == department_id).first()
  if not department:
    raise HTTPException(status_code=404, detail="Department not found")
  return department

if __name__ == "__main__":
  # 直接在代码中启动uvicorn服务器
  uvicorn.run(
    app="fastapi_sqlalchemy:app",    # 指定要运行的FastAPI应用实例
    host="0.0.0.0", # 允许外部访问（本地可通过127.0.0.1或localhost访问）
    port=8000,   # 端口号
    reload=True  # 开发模式：代码修改后自动重启（生产环境需去掉）
)