Python大模型基础
摘要：本文是一份全面的 Python 后端开发技术笔记，系统梳理了从 Python 基础语法（注释、变量、数据类型、流程控制、容器类型、函数、文件操作）到面向对象编程（封装、继承、多态），再到高级特性（迭代器、生成器、闭包、装饰器、内存管理）的完整知识体系，并深入讲解了进程、线程与协程的并发编程模型。文章还覆盖了 Linux 常用命令、Docker 容器化部署（镜像、容器、数据卷、Docker Compose）、FastAPI 高性能 Web 框架，以及 SQLAlchemy ORM 数据库操作（模型定义、增删改查、关联查询）等后端开发核心技能，适合作为系统复习与实战参考。
一. python
1.python应用
Web应用开发(FastAPI、Django、Flask) 网络爬虫 系统网络运维 数据分析与科学计算(Numpy、Scipy) 人工智能与机器学习(TensorFlow、Keras)
2. 基础知识
2.1 注释
对代码解释说明 提高代码可读性 屏蔽掉暂时不需要的代码 定位程序出错的位置
2.1.1 注释分类
单行注释（行注释） 多行注释（块注释）
2.2 变量
2.2.1 标识符命名规则
只能包含字母、数字和下划线，且不能以数字开头 区分大小写 不要和关键字重复
2.3 进制及转换
-
二进制：以0b或0B开头表示 bin(dec)
-
八进制：以0o开头表示 oct(dec)
-
十六进制：以0x或0X开头表示，此处的A-F不区分大小写 hex(dec)
2.4 数据类型
-
基本数据类型
-
数值(整数（int）、浮点数（float）、复数（complex）、布尔（bool）)
-
字符串（str）
-
容器数据类型
列表（list） 元组（tuple）集合（set）字典（dict）
-
特殊数据类型
None : 表示空值或缺失值，只有一个值 None。常用于函数没有返回值时，或者表示变量没有被赋值。
-
不可变数据（3 个）：Number（数字）、String（字符串）、Tuple（元组）
-
可变数据（3 个）：List（列表）、Dictionary（字典）、Set（集合）
2.4.1 int整型
num1 = True num2 = 10 print(type(num1)) # <class 'bool'> print(type(num2)) # <class 'int'> print(type(num1) == type(num2)) # False print(isinstance(num1, bool)) # True print(isinstance(num1, int)) # True，Python3中，bool是int的子类 print(isinstance(num2, int)) # True
2.4.2 float浮点型
2.4.3 bool布尔型
False None 0 0.0 所有的空容器（空列表、空元组、空字典、空集合、空字符串）
2.4.4 str 字符串
3. 流程控制语句
顺序 分支(单分支/多分支/嵌套分支) match case语句 三目运算符(表达式1 if 判断条件 else 表达式2) while 循环 for 循环(遍历可迭代对象，如列表或字符串)
-
continue(跳过当前循环块中的剩余语句，继续进行下一轮循环)
-
break(for 或while循环终止)
4.容器数据类型
4.1 序列（Sequence）
-
一种基本且核心的数据结构，它允许我们以有序的方式存储和操作数据。序列可以包含不同类型的元素，并且支持通过索引来访问和修改这些元素。
-
常见的序列类型包括：列表（List）、元组（Tuple）、字符串（String）
4.2 列表List
可变有序的元素集合 用 [] 定义 索引或下标访问 元素可以是不同的类型
-
访问列表
通过索引获取列表中元素 列表切片 向列表中添加元素(append/insert/extend) 修改列表中元素(下标/切片)
-
列表推导式
squares = [x**2 for x in range(10) if x % 2 == 0]
-
zip() 函数可将多个可迭代对象中对应元素打包为一个个元组
4.3 字符串 str
替换 str.replace(old,new[,max]) 分割str.split(x) 合并(x作为分隔符)x.join(seq)
4.4 元组 Tuple
不可变有序的元素集合,不能对元组中的元素进行修改操作 使用 () 定义 索引或下标访问 元素可以是不同的类型
4.5 集合Set
无序没有索引且不包含重复元素 用 {} 定义，数据之间使用 , 分隔，也可以使用set()定义 元素可以是不同的类型 可以进行数学上的集合操作，如并集、交集和差集 适用于需要快速成员检查、消除重复项和集合运算的场景
-
添加元素.add()
-
删除元素.remove()
-
遍历集合 in
4.6 字典Dictionary
无序没有索引的键值对集合，键是唯一的，而值可以重复 用 {}或dict() 定义，键（key）和值（value）使用 : 连接，通过键来获取对应的值 值可以取任何数据类型，但键必须是不可变的，如字符串、数字、元组
5. 函数
-
定义函数使用 def 关键字 先定义再调用
-
若函数的形参是定长参数，可以通过 * 和 ** 对列表、元组、字典等解包传参
-
/ 前的参数必须使用位置传参，* 后的参数必须用关键字传参
-
global : 修改全局变量
-
nonlocal : 内部作用域修改外部作用域(函数中)的变量
-
用 lambda 来定义匿名函数 (lambda 参数列表: 表达式)
def operator(a, b): return a + b def function(a, b, operator): return operator(a, b) print(function(1, 2, lambda x, y: x + y))
6. 文件操作
-
纯文本文件
-
二进制文件(如图片文件（jpg、png），视频文件（avi）等)
r : 读 ; w : 写 ; a : 追加写 ; x : 创建文件写 ; b : 二进制 ; t : 文本模式 ; + : 读写
-
read([size]) 可以从文件中读取数据，size 表示要从文件中读取的数据的长度（编码模式b：单位是字节，编码模式t：单位是字符），如果没有传入 size 则读取文件中所有的数据
-
readline([size]) 可以从文件中读取整行数据，也可以通过 size 设置读取数据的长度
-
readlines([size]) 读取所有行并返回列表，若给定 size>0，返回总和大约为 size 字节的行， 实际读取值可能比 size 大
6.1 with上下文管理器
try:
with open("test.txt", "w") as f:
f.write(a)
finally:
print("文件是否关闭：", f.closed) # 文件是否关闭： True
7. 面向对象之类和对象
-
类是创建对象的模板 对象是类的实例（Instance）
-
类名一般使用大驼峰命名法
-
类方法在类中通过 @classmethod 定义，第一个参数为cls，代表类本身
-
静态方法在类中通过 @staticmethod 定义 , 不访问实例属性或类属性，只依赖于传入的参数
-
魔法方法 : 方法名中有两个前缀下划线和两个后缀下划线的方法
8. 面向对象之三大特性
8.1 封装
-
将变量和函数写入类中的操作即为封装，即类中封装了属性和方法
-
通过双下划线定义私有属性或私有方法
8.1.1 property
-
方法转换为属性 : 可通过@property装饰器将一个方法转换为属性来调用。转换后可直接使用 .方法名 来使用，而无需使用 .方法名()
class Person:
def __init__(self, name):
self.name = name
@property
def eat(self):
print(f"{self.name} is eating...")
p = Person("张三")
p.eat # 张三 is eating...
-
只读属性 : 将方法名设置为去掉双下划线的私有属性名，方法中返回私有属性
class Person:
def __init__(self, name):
self.__name = name
@property
def name(self):
return self.__name
p = Person("张三")
print(p.name) # 张三
p.name = "李四" # 报错
-
读写属性:将方法名设置为去掉双下划线的私有属性名，方法中返回私有属性
class Person:
def __init__(self, name):
self.__name = name
@property
def name(self):
return self.__name
@name.setter
def name(self, name):
self.__name = name
p = Person("张三")
print(p.name) # 张三
p.name = "李四"
print(p.name) # 李四
8.2 继承
-
子类继承父类中的属性和方法，实现代码重用。子类可以新增自己特有的方法，也可以重写父类的方法
-
子类不能继承父类的私有属性和私有方法
-
支持多继承
-
子类可以在类中使用 super().方法名() 或 父类名.方法名() 来调用父类的方法
8.2.1 方法解析顺序（mro）
-
类名.mro 访问类的继承链来查看方法解析顺序
8.2.2 方法重写
-
在子类中定义与父类方法重名的方法，调用时会调用子类中重写的方法
8.3 多态
9. 错误和异常
-
语法错误 : 程序解析时遇到的错误
-
异常 : 运行期检测到的错误
-
用 try except 语句来捕获异常并处理
-
用 raise 来抛出异常
-
assert用于判断一个表达式，在表达式条件为False的时候触发异常，常用于调试程序
def int_add(x, y):
assert isinstance(x, int) and isinstance(y, int), "参数类型错误"
return x + y
print(int_add(1, 2)) # 3
print(int_add("1", "2")) # AssertionError: 参数类型错误
10. 模块和包
-
使用from import *导入模块时，在被导入的模块中使用 all设置哪些内容可以被导入。all 的设置只针对使用 from import * 导入模块时有效
__all__ = ["num","add"] # 内容必须要用引号引起来 num = 100 num1 = 200 _str1="abc" def add(a, b): """求两个数的和""" return a + b
10.1 常见的标注库(包)
-
os : 多种操作系统接口
-
sys : 系统相关的形参和函数
-
time : 时间的访问和转换\
-
datetime : 操作日期和时间的类
-
math : 数学函数
-
random : 生成伪随机数
-
re : 正则表达式匹配操作
-
json
-
collections :
10.2 引入第三方库
-
阿里云：Simple Index
-
清华大学：Simple Index
-
pip install 包名
-
临时使用其他源 : pip install -i Simple Index 包名
-
永久修改源 : pip config set global.index-url Simple Index
打包自己的库并安装
-
pip install setuptools
-
在包外创建一个 setup.py文件
-
from distutils.core import setup setup( name="graphic", # 需要打包的名字 version="1.0", # 版本 py_modules=["graphic.circle", "graphic.rectangle"], # 需要打包的模块 )
-
pip install setuptools
-
11.Python高级语法
11.1 浅拷贝与深拷贝
-
直接赋值：对象的引用（别名），不产生拷贝
-
浅拷贝：拷贝父对象，不会拷贝对象的内部的子对象。拷贝后只有第一层是独立的
-
深拷贝：完全拷贝了父对象及其子对象。拷贝后所有层都是独立的
-
使用 copy() 对其浅拷贝。使用 id() 查看列表地址和列表中各个元素的地址
-
使用 copy 模块的 deepcopy() 函数
11.2 迭代器
能被 next() 一个一个取值的对象 __iter__()：返回迭代器本身 __next__()：返回下一个值
-
可迭代对象(可以直接作用于 for 循环的数据类型)有以下几种：
-
容器，如 list 、 tuple 、 dict 、 set 、 str 等。
-
generator ，包括生成器和带 yield 的generator function
-
判断是否是可迭代对象（Iterable)
from collections.abc import Iterable
print(isinstance([], Iterable)) # True
print(isinstance((), Iterable)) # True
print(isinstance(set(), Iterable)) # True
print(isinstance({}, Iterable)) # True
print(isinstance("100", Iterable)) # True
print(isinstance(100, Iterable)) # False
-
判断是否是迭代器（Iterator）
from collections.abc import Iterator
print(isinstance([], Iterator)) # False
print(isinstance((), Iterator)) # False
print(isinstance(set(), Iterator)) # False
print(isinstance({}, Iterator)) # False
print(isinstance("100", Iterator)) # False
print(isinstance((x for x in range(10)), Iterator)) # True
list = [1, 2, 3]
it = iter(list) # 创建迭代器对象
print(next(it)) # 输出迭代器的下一个元素,1
print(next(it)) # 2
print(next(it)) # 3
print(next(it)) # StopIteration
list = [1, 2, 3]
it = iter(list) # 创建迭代器对象
for i in it:
print(i)
11.3 生成器
-
一个用于创建迭代器的简单而强大的工具。函数里有
yield，是生成器函数，yield是暂停而不是返回，下次next()会从暂停处继续执行-
列表推导式：一次性生成所有数据 生成器表达式：惰性生成
-
大模型流式输出（SSE）本质就是生成器模式——模型每生成一个 token，前端就收到一段文本
-
-
使用推导式创建生成器
generator = (x for x in range(5)) # 创建生成器 print(generator) # <generator object <genexpr> at 0x0000026C2066CB80> for x in generator: print(x)
-
使用函数创建生成器
def fibo(): # 斐波那契数列 a, b = 0, 1 while True: yield b a, b = b, a + b f = fibo() print(next(f)) # 1 print(next(f)) # 1 print(next(f)) # 2 print(next(f)) # 3 print(next(f)) # 5
11.4 闭包
内部函数引用了外部函数的变量，形成了 “变量捕获”，使得变量不会随外部函数的结束而销毁，而是被内部函数 “保留” 在内存中，直到内部函数被销毁 读取不需要声明，修改必须用nonlocal
-
闭包的核心价值是 “保留状态” 和 “封装数据”，常见应用场景包括:
-
装饰器(不修改原函数代码的前提下，添加额外功能（如日志、计时）)
-
数据封装与隐藏(模拟类的私有变量)
-
函数工厂(动态生成具有不同参数的函数，避免重复代码)
-
-
闭包会持有外部变量的引用，如果引用的是大对象且未及时释放，会导致内存泄漏
- 不再使用时手动解除引用,设为None（触发垃圾回收）
11.5 装饰器
-
不修改原有函数代码的基础上，动态地增加或修改函数的功能
from math import sqrt
def func(x): """开根号""" return sqrt(x)
def decorator(f): def inner(x): x = abs(x) # 求x的绝对值 return f(x)
return inner
func = decorator(func) print(func(-4)) # 2.0
from math import sqrt
def decorator(f): def inner(x): x = abs(x) # 求x的绝对值 return f(x)
return inner
@decorator def func(x): """开根号""" return sqrt(x)
print(func(-4)) # 2.0
11.6 内存管理与垃圾回收
Python 内存由私有堆空间管控，所有对象与数据结构置于此，不能直接访问，内存管理器负责堆空间内存分配。其有内置垃圾收集器，借助对象引用计数,循环引用,分代回收机制，回收未用内存释放给堆空间
12. 进程与线程
-
并发 : 单个 CPU 处理多个任务。各个任务交替执行一段时间
-
并行 : 多个 CPU 同时执行多个任务
12.1 多进程
-
进程是操作系统进行资源分配的基本单位
-
操作系统中一个正在运行的程序或软件就是一个进程
-
每个进程都有自己独立的一块内存空间
12.1.1 创建进程
-
使用Process()创建进程
-
自定义Process子类创建进程,继承Process
-
当需要启动大量子进程时，可以使用进程池Pool()
12.2 多线程
-
线程是处理器任务调度和执行的基本单位
-
一个进程至少有一个线程，也可以运行多个线程
-
多个线程之间可共享数据
-
线程运行出错异常后，如果没有捕获，可能会导致整个进程崩溃
-
多线程是指在同一进程中同时执行多个任务
12.2.1 创建进程
-
使用Thread创建线程
-
自定义Thread子类创建线程,继承Thread
-
线程池创建ThreadPoolExecutor()
13. 协程（Coroutine）
-
Python中处理高并发场景的强大工具，单线程非阻塞，解决IO密集型任务的阻塞效率问题，事件循环+协程,底层依托生成器的暂停/恢复机制和操作系统的非阻塞IO/IO多路复用。一种用户态的轻量级线程，它不像线程那样由操作系统调度，而是由程序自身控制，所以也叫微线程，事件循环作为总调度
13.1 协程的优势
-
调度者：由程序自身进行调用，主动让出CPU给其他协程（通过 await 关键字）
-
轻量级：创建成本极低，一个线程程可以轻松创建数万个协程
-
无锁机制：同一时间只有一个协程运行，无需锁保护共享资源
-
与线程相比，协程避免了操作系统级别的上下文切换开销，只有极小的函数调用开销，在处理大量 I/O 操作时效率更高
13.2 async和await
13.2.1 定义协程函数：async def
-
使用async def关键字定义的函数就是协程函数，它返回一个协程对象
import asyncio
这是一个协程函数
async def hello():
print("Hello")
# await 关键字只能在协程函数内部使用
await asyncio.sleep(1) # 模拟I/O操作
print("World")
调用 hello() 不会立即执行，只会返回一个协程对象
coro_obj = hello()
print(type(coro_obj)) # 输出: <class 'coroutine'>
13.3 运行协程：事件循环
import asyncio
import time
async def my_func(name, delay):
print(f"任务 {name} 开始")
# 遇到 await，协程让出控制权
await asyncio.sleep(delay)
print(f"任务 {name} 完成")
return f"任务 {name} 结果"
async def main_sync():
print("--- 串行执行（总耗时 1s + 2s = 3s）---")
# await 串行执行：必须等待前一个 task 完成，才会开始下一个
await my_func("A", 1)
await my_func("B", 2)
if name == "main":
start_time = time.time()
asyncio.run(main_sync())
print(f"串行耗时: {time.time() - start_time:.2f}秒")
输出:
--- 串行执行（总耗时 1s + 2s = 3s）---
任务 A 开始
任务 A 完成
任务 B 开始
任务 B 完成
串行耗时: 3.02秒
13.4 事件循环(Event Loop) 详述
-
事件循环是协程并发模型的核心，它是负责调度、管理和执行所有协程的“心脏”
13.4.1 注册任务
-
asyncio.run(main_coro)(顶级注册，启动程序)
-
asyncio.create_task(coro)(最常用，实现并发)
-
await coro(隐式注册，实现串行)
13.5 实现并发：Task与asyncio.gather()
import asyncio import time
async def my_func(name, delay): print(f"任务 {name} 开始...") await asyncio.sleep(delay) print(f"任务 {name} 完成.") return f"结果 of {name}"
async def main_concurrent(): print("--- 并发执行（总耗时 max(1s, 2s) = 2s）---")
# 1. 创建 Task：协程被包装成任务，立即启动在后台运行
task1 = asyncio.create_task(my_func("C", 1))
task2 = asyncio.create_task(my_func("D", 2))
# 2. 等待 Task 完成：await asyncio.gather() 等待所有任务结束
# 注意：这里的 await 只是等待结果，任务本身早已开始并发运行。
results = await asyncio.gather(task1, task2)
print(f"所有结果: {results}")
if name == "main": start_time = time.time() asyncio.run(main_concurrent()) print(f"并发总耗时: {time.time() - start_time:.2f}秒") 输出: --- 并发执行（总耗时 max(1s, 2s) = 2s）--- 任务 C 开始... 任务 D 开始...
二、Linux
-
Linux系统中一切皆文件
-
APT常用命令：
list - 根据名称列出软件包
search - 搜索软件包描述
show - 显示软件包细节
install - 安装软件包
reinstall - 重新安装软件包
remove - 移除软件包
update - 更新可用软件包列表
1.常用基本命令
帮助命令: man ls - 查看ls命令的帮助信息 help cd - 查看cd命令的帮助信息 常用快捷键: Ctrl + L - 清屏；彻底清屏是：clear Ctrl + U - 清除当前敲的命令 文件命令： pwd - 显示当前工作目录的绝对路径 touch test/a/note.txt - 创建空文件 cp -r test/b/ ./ - 递归复制 rm -r b/ - 递归删除 mv a.txt b.txt - 重命名 mv test/a.txt ./ - 移动文件 cat -n a.txt - 查看文件显示行号 tail - 输出文件尾部内容(默认10行; -n 5:尾部5行;-F:实时输出) echo - 输出内容到控制台 vim命令： yy - 复制光标当前一行 p - 箭头移动到目的行粘贴 u - 撤销上一步 dd - 删除光标当前行 :wq! - 强制保存退出 用户管理命令： su 用户名称 - 切换用户，只能获得用户的执行权限，不能获得环境变量 用户管理命令： su 用户名称 - 切换用户，只能获得用户的执行权限，不能获得环境变量 su - 用户名称 - 切换到用户并获得该用户的环境变量及执行权限 useradd [-g 组名] 用户名 - 添加新用户到某个组 passwd 用户名 - 设置用户密码 id 用户名 - 查看用户是否存在 cat /etc/passwd - 查看创建了哪些用户 userdel 用户名 - 删除用户但保存用户主目录 userdel -r 用户名 - 删除用户的同时，删除与用户相关的所有文件 文件权限： 0 123 456 789 类型(d:目录/l:链接文档) 属主(rwx:读写执行) 属组 其他用户权限 chmod 改变权限： u:所有者 g:所有组 o:其他人 a:所有人(u、g、o的总和)rwx=4+2+1=7 chmod u-x,o+x passwd - 修改文件属主用户执行权限,并使其他用户具有执行权限 chmod -R 777 test - 压缩c：tar -zcvf xxx.tar.gz 解压x：tar -zxvf xxx.tar.gz 查看磁盘空间:df -h
三.Docker
官网地址：Docker: Accelerated Container Application Development
-
提供了一套标准化的环境打包与运行方案：把代码、依赖、配置、系统环境全部打包在一起，形成一个可移植的容器
添加国内镜像源 "registry-mirrors": [ "https://docker.xuanyuan.me", "https://docker.1ms.run", "https://docker.m.daocloud.io", "http://hub-mirror.c.163.com", "https://docker.unsee.tech", "https://dockerpull.org", "https://docker.1panel.live", "https://dockerhub.icu", "https://docker.nju.edu.cn", "https://registry.docker-cn.com", "https://docker.mirrors.ustc.edu.cn", "https://mirror.baidubce.com", "https://5tqw56kt.mirror.aliyuncs.com", "https://docker.hpcloud.cloud", "http://mirrors.ustc.edu.cn", "https://docker.chenby.cn", "https://docker.ckyl.me", "http://mirror.azure.cn", "https://hub.rat.dev", "https://mbirror.ccs.tencentyun.com" ] Docker部署: #更新软件包 sudo apt update sudo apt upgrade #安装docker依赖 sudo apt install software-properties-common sudo apt-get install ca-certificates curl gnupg lsb-release #添加Docker官方GPG密钥 curl -fsSL http://mirrors.aliyun.com/docker-ce/linux/ubuntu/gpg | sudo apt-key add - #添加Docker软件源（输入后根据提示按Enter） sudo add-apt-repository "deb [arch=amd64] http://mirrors.aliyun.com/docker-ce/linux/ubuntu $(lsb_release -cs) stable" #安装docker（输入后根据提示输入 y ） sudo apt-get install docker-ce docker-ce-cli containerd.io --fix-missing #安装完毕，启动docker，并查看状态 sudo systemctl start docker sudo systemctl status docker
1.Docker优势
-
可移植性
-
可伸缩性
-
隔离性
2.Docker核心概念
2.1 镜像（Image）—类比「软件安装包」
-
只读（不能直接修改），可作为模板，创建多个容器
-
包含了运行某个应用程序所需的全部环境（比如操作系统、软件依赖、配置文件等）
2.2 容器（Container）—类比「已安装好的软件」
-
镜像的运行实例，是一个可读写的独立环境，相当于“把安装包安装好后，能直接使用的软件”
-
可启动、停止、删除，多个容器相互独立；容器删除后，里面的数据默认会丢失（可通过“数据卷”保留，后续会讲）
2.3 仓库（Repository）—类比「手机应用商店」
官方仓库地址：https://hub.docker.com/
-
用于存放和下载镜像的平台
2.4 数据卷（Volume）—— 类比「电脑的D盘（专门存数据）」
-
数据卷是Docker中用于持久化数据的工具，相当于“给容器挂一个独立的存储盘”，容器删除后，数据卷中的数据不会丢失
2.5 核心工具Dockerfile和Docker Compose
2.5.1 Dockerfile —自定义镜像的“构建脚本”
-
一个纯文本文件（无后缀名），包含一系列简单的指令，用于告诉Docker“如何构建自定义镜像”
2.5.2 Docker Compose —多容器的“一键管理工具”
-
一个用于管理多容器的工具，通过一个yaml格式的配置文件（固定命名为docker-compose.yml），定义多个关联容器的配置（比如镜像、端口、依赖关系），执行一条命令就能一键启动、停止所有容器
# 核心功能：一键启动MySQL容器+Python应用容器，实现容器间通信、数据持久化 services: # 容器1：MySQL服务容器（命名为mysql-db，容器间可通过该名称访问） mysql-db: # 基础镜像：指定MySQL 8.0.45版本（稳定性高，适配Python连接） image: mysql:8.0.45 # 重启策略：容器意外关闭/服务器重启时，自动重启（保证服务可用性） restart: always # 环境变量：配置MySQL核心参数（无需手动进入容器修改） environment: MYSQL_ROOT_PASSWORD: "123456" # root用户密码（自定义，Python连接需一致） MYSQL_DATABASE: "test_db" # 自动创建初始数据库（无需手动建库） # 字符编码配置：适配MySQL 8.0+，避免中文乱码/连接报错 MYSQL_INITDB_ARGS: "--character-set-server=utf8mb4 --collation-server=utf8mb4_unicode_ci" # 端口映射：本地9999端口 → 容器内3306端口（避免与本地MySQL端口冲突） ports: - "9999:3306" # 数据卷挂载：Windows本地D盘mysql-data目录 → 容器内MySQL数据目录（持久化数据） volumes: - D:/mysql-data:/var/lib/mysql # 加入自定义网络：与Python容器互通（容器间通信的基础） networks: - app-network # 健康检查：判断MySQL服务是否真正就绪（而非仅容器启动），解决连接拒绝问题 healthcheck: test: ["CMD", "mysqladmin", "ping", "-h", "localhost", "-uroot", "-p123456"] # 检测MySQL是否可ping通 interval: 3s # 每3秒检查一次 timeout: 3s # 单次检查超时时间3秒 retries: 10 # 最多重试10次（覆盖MySQL启动初始化时间） start_period: 5s # 容器启动后，延迟5秒再开始检查（避免过早检测） # 容器2：Python应用容器（命名为python-app，业务代码运行载体） python-app: # 构建规则：基于当前目录下的Dockerfile，构建自定义Python镜像 build: . # 依赖条件：等待mysql-db容器的健康检查通过后，再启动本容器（核心修复：解决MySQL未就绪问题） depends_on: mysql-db: condition: service_healthy # 仅当MySQL服务就绪时，才启动Python容器 # 环境变量：传递MySQL连接信息（Python代码可通过os.getenv获取，无需硬编码） environment: MYSQL_HOST: "mysql-db" # 容器间通信地址：直接填MySQL容器名（无需IP） MYSQL_PORT: "3306" # MySQL容器内端口（非本地9999） MYSQL_USER: "root" # 与MySQL容器的用户名一致 MYSQL_PASSWORD: "123456" # 与MySQL容器的密码一致（修改需同步） MYSQL_DB: "test_db" # 与MySQL容器自动创建的数据库名一致 # 加入自定义网络：与MySQL容器在同一网络，才能通信 networks: - app-network # 自定义网络：桥接模式（Docker默认），保证两个容器相互可见、可访问 networks: app-network: driver: bridge
3.基础命令
docker --version docker compose version docker pull mysql:8.0.45 查看镜像 docker images ocker run -d --name mysql-db -p 9999:3306 -e MYSQL_ROOT_PASSWORD=123456 mysql:8.0.45 查看容器 docker ps
四.FastAPI
-
一个现代、快速（高性能）的Web框架，用于构建API
-
类型提示（type hints）和异步编程（asyncio）能力，使得代码易于编写、阅读和维护
-
有自动交互式文档（基于 OpenAPI 规范和 JSON Schema）、数据验证、依赖注入（Dependency Injection）等功能
from fastapi import APIRouter
# 实例化一个 APIRouter，可指定前缀（所有路由自动加上 /users）
router = APIRouter(
prefix="/users",
tags=["用户管理"] # 文档中归类为「用户管理」
)
# 定义用户相关路由
@router.get("/")
def get_all_users():
return {"message": "获取所有用户列表"}
from fastapi import APIRouter
router = APIRouter(
prefix="/items",
tags=["商品管理"] # 文档中归类为「商品管理」
)
# 定义商品相关路由
@router.get("/")
def get_all_items():
return {"message": "获取所有商品列表"}
from fastapi import FastAPI
from routers import user, item # 导入子模块路由
import uvicorn
app = FastAPI(title="路由分发示例")
# 挂载用户路由：所有 /users 开头的请求由 user.router 处理
app.include_router(user.router)
# 挂载商品路由：所有 /items 开头的请求由 item.router 处理
app.include_router(item.router)
# 主应用自身也可以定义路由
@app.get("/")
def root():
return {"message": "欢迎访问主页面"}
if __name__ == "__main__":
# 直接在代码中启动uvicorn服务器
uvicorn.run(
app="main:app", # 指定要运行的FastAPI应用实例
host="0.0.0.0", # 允许外部访问（本地可通过127.0.0.1或localhost访问）
port=8000, # 端口号
reload=True # 开发模式：代码修改后自动重启（生产环境需去掉）
)
五. SQLAlchemy
1.ORM
-
ORM（Object-Relational Mapping，对象关系映射）是一种编程技术，它将数据库中的表结构映射为编程语言中的类，将表中的行映射为类中的对象，让开发者可以用面向对象的方式操作数据库，而无需直接编写 SQL 语句。
-
屏蔽数据库差异
-
同一套代码可适配多种数据库（MySQL、PostgreSQL 等），无需修改核心逻辑
-
-
简化开发流程
-
用类、对象、方法替代 SQL 语句，降低数据库操作的学习成本
-
-
提高代码可读性
-
将数据库操作与业务逻辑融合，代码更符合面向对象思维
-
-
自动处理类型转换
-
无需手动转换数据库字段与Python类型（如MySQL的INT与Python的int）
-
2.SQLAlchemy
SQLAlchemy的架构分层可分为DBAPI 层、SQLAlchemy Core（核心层）、SQLAlchemy ORM（对象关系映射层）
2.1 DBAPI（数据库通信层）
-
Python数据库接口规范，是SQLAlchemy与底层数据库通信的 “桥梁”。
-
pymysql 是 MySQL 的 DBAPI，psycopg2 是 PostgreSQL 的 DBAPI
2.2 SQLAlchemy Core（核心层）
SQLAlchemy的 “基础工具集”，提供了SQL表达式语言、数据库连接管理等核心能力，即使不使用 ORM，也能通过 Core 操作数据库
2.2.1 Schema / Types
-
定义数据库的模式（Schema）和数据类型（Types）
-
Schema对应数据库的表、列、约束等结构（比如定义一张表有哪些字段、字段类型是什么）。Types封装了数据库支持的数据类型（如 Integer、String、DateTime 等），并提供 Python 类型与数据库类型的映射
2.2.2 SQL Expression Language
-
用Python代码生成SQL语句的 “表达式语言”。
它允许你用面向对象的方式编写 SQL（比如用 table.c.column == value 表示 WHERE column = value），既保留了 SQL 的灵活性，又能避免手写 SQL 带来的语法错误和安全问题（如 SQL 注入）
2.2.3 Engine
管理数据库连接的 “引擎”，是与数据库交互的 “入口”。
Engine 负责创建和维护数据库连接，还会集成连接池（Connection Pooling）和方言（Dialect）
2.2.4 Connection Pooling
管理数据库连接池，提升数据库操作性能。
连接池会预先创建一批数据库连接并复用，避免频繁创建 / 销毁连接的开销，尤其在高并发场景下能显著提高效率
2.4.5 Dialect
处理 “方言” 差异，适配不同数据库的 SQL 语法和特性。
不同数据库（如 MySQL 和 PostgreSQL）的 SQL 语法、函数可能有差异（比如 MySQL 的 LIMIT 和 PostgreSQL 的 LIMIT/OFFSET 用法不同）。Dialect 会对这些差异做 “翻译”，让上层代码能以统一的方式操作不同数据库
2.3 SQLAlchemy ORM（对象关系映射层）
在Core的基础上，提供象关系映射（ORM）能力，让开发者可以用 “类和对象” 的方式操作数据库（比如用 User 类对应 users 表，用 user = User(name="Alice") 表示新增一条用户记录）。
ORM本质是对Core的 “封装”—— 它会把面向对象的操作（如创建对象、查询对象）自动转换为Core能理解的 SQL 表达式通过，最终 Engine 执行。这样开发者可以更聚焦于业务逻辑，而无需关注底层SQL实现。
2.4 核心组件
-
引擎（Engine）：
Engine是与数据库的连接入口，负责管理连接池和执行 SQL 语句 -
基类（Base）：所有 ORM 模型的父类，用于统一管理数据库表结构
-
会话（Session）：用于执行数据库操作的会话对象，负责暂存、提交、回滚数据操作
2.5 使用案例
2.5.1 定义Base基类
from sqlalchemy.ext.declarative import declarative_base # 生成基类，所有模型需继承该类 Base = declarative_base()
2.5.2 定义ORM模型类（映射MySQL表）继承Base基类
from sqlalchemy import Column, Integer, String, ForeignKey, Date
from sqlalchemy.orm import relationship
from demo.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" # 可选：立即加载部门数据，减少查询次数
)
2.5.3 创建引擎以及会话
from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker DATABASE_URL = "mysql+pymysql://root:123456@localhost:3306/fastapi_db" engine = create_engine( DATABASE_URL, echo=True, # 是否启用日志输出 开发环境启用，生产环境关闭 pool_pre_ping=True # 连接前检查有效性，避免连接失效 ) SessionLocal = sessionmaker( autocommit=False, # 关闭自动提交，需手动 commit() bind=engine )
2.5.4 建表
rom demo.base import Base from demo.database import engine from demo.models import Employee,Department #必须要加 # 创建表 def create_table(): # 创建所有模型对应的表 Base.metadata.create_all(bind=engine)
2.5.5 新增数据（Create）
def insert_data(): db = SessionLocal() try: new_dept = Department(name="研发部",location="北京总部") db.add(new_dept) db.commit() db.refresh(new_dept) # 刷新对象，获取自增的 id 等字段 emp1 = Employee(name="张三",age=30,hire_date=date(2023, 1, 1),department_id=new_dept.id) emp2 = Employee(name="李四",age=28,hire_date=date(2023, 3, 15),department_id=new_dept.id) db.add_all([emp1, emp2]) db.commit() # 刷新员工对象，获取自增 ID db.refresh(emp1) db.refresh(emp2) except Exception as e: db.rollback() # 出错时回滚 finally: db.close() # 关闭会话
2.5.6 删除数据（Delete）
def delete_data(): session = SessionLocal() try: emp = session.query(Employee).filter(Employee.name == "李四").first() if emp: session.delete(emp) session.commit() except Exception as e: session.rollback() # 出错时回滚 finally: session.close() # 关闭会话
2.5.7 修改数据（Update）
def update_data(): session = SessionLocal() try: emp = session.query(Employee).filter(Employee.name == "张三").first() if emp: emp.age = 31 # 直接修改属性 session.commit() # 提交更新 except Exception as e: session.rollback() # 出错时回滚 finally: session.close() # 关闭会话
2.5.8 查询数据（Read）
def read_data():
session = SessionLocal()
try:
dept = session.get(Department,1)
rd_employees = session.query(Employee).filter(
Employee.department_id == 1 # 按部门ID过滤
).all()
old_employees = session.query(Employee).filter(Employee.age > 30).all()
from sqlalchemy import and_, or_
emp = session.query(Employee).filter(
and_(Employee.age.between(30, 40), Employee.department_id == 1)
).first()
emps = session.query(Employee).filter(
or_(Employee.department_id == 2, Employee.age > 32)
).all()
result = session.query(Employee, Department).join(
Department, Employee.department_id == Department.id
).all()
from sqlalchemy.orm import joinedload
employees = session.query(Employee).options(
joinedload(Employee.department) # 预加载关联的 department
).all()
from sqlalchemy import func
dept_emp_count = session.query(
Employee.department_id,
func.count(Employee.id).label("count") # 别名 count
).group_by(Employee.department_id).subquery() # 转为子查询
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的部门
locations = session.query(Department.location).join(
Employee
).distinct().all() # distinct() 去重
first_emp = session.query(Employee).first()
all_depts = session.query(Department).all()
except Exception as e:
session.rollback() # 出错时回滚
finally:
session.close() # 关闭会话
2.6 sqlacodegen通过表生成类
# pip install sqlacodegen
# pip install pymysql # MySQL 驱动
# CREATE TABLE departments (id INT PRIMARY KEY AUTO_INCREMENT,name VARCHAR(50) NOT NULL UNIQUE COMMENT '部门名称',location VARCHAR(100) COMMENT '部门位置',created_at DATETIME DEFAULT CURRENT_TIMESTAMP);
# CREATE TABLE employees (id INT PRIMARY KEY AUTO_INCREMENT,name VARCHAR(50) NOT NULL COMMENT '员工姓名',age INT COMMENT '年龄',hire_date DATE COMMENT '入职日期',department_id INT COMMENT '所属部门ID',FOREIGN KEY (department_id) REFERENCES departments(id) ON DELETE SET NULL,INDEX idx_dept (department_id) -- 部门索引，优化查询);
import subprocess
import sys
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from demo1.table_2_models import Departments, Employees
url = f"mysql+pymysql://{db_user_name}:{db_password}@{db_host}:{db_port}/{db_name}?charset=utf8mb4"
engine = create_engine(url, echo=True)
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
cmd = [venv_python, "-m", "sqlacodegen", url]
result = subprocess.run(cmd, capture_output=True, text=True, encoding="utf-8")
with open(output_path, "w", encoding="utf-8") as f:
f.write(result.stdout)
def insert_dept_emp():
emp = Employees(id=100,name='zs',age=20,hire_date='2025-10-10')
dept = Departments(id=10,name='研发部',location='北京',created_at='2025-10-10',employees=[emp])
with Session(engine) as session:
session.add(dept)
try:
session.commit()
except Exception as e:
session.rollback()更多推荐
4
0- 0
已为社区贡献3条内容
所有评论(0)