Python编程
Python 编程：从入门到运维实战
目录
- Python 简介与环境部署
- 基本使用：交互式与脚本模式
- 变量与数据类型
- 运算符
- 流程控制：if/elif/else
- for 循环
- while 循环
- 循环控制进阶
- 数据类型深入
- 函数
- switch 语句替代方案
- 模块与代码封装
- 反射机制
- 正则表达式
- 文件 IO 与操作系统交互
- 异常处理
- 装饰器
- 面向对象编程
- MySQL 数据库操作
1. Python 简介与环境部署
1.1 Python 是什么？
类比：如果说 C 语言像手动挡汽车——功能强大但操作复杂，那 Python 就像自动挡汽车——同样能到达目的地，但驾驶体验轻松得多。
Python 是一种解释型、面向对象的高级编程语言，诞生于 1989 年，由荷兰程序员 Guido van Rossum 在圣诞节期间开发。名字来源于他喜爱的喜剧团体 "Monty Python"。
Python 在运维中的地位：
- Google 用 Python 编写网页爬虫和搜索引擎组件
- NASA 在多个系统中使用 Python 作为脚本语言
- YouTube 的大部分服务由 Python 编写
- 几乎所有 Linux 发行版都预装了 Python
1.2 Python 2 vs Python 3
| 特性 | Python 2 | Python 3 |
|---|---|---|
| 默认编码 | ASCII | UTF-8 |
| print 语句 | print "hello" | print("hello") |
| 整数除法 | 3/2 = 1 | 3/2 = 1.5 |
| 官方支持 | 2020年已停止维护 | 持续更新中 |
| 字符串 | str 是字节串 | str 是 Unicode |
1.3 环境安装
Linux 源码安装（以 CentOS 为例）：
# 安装依赖
[root@localhost ~]# yum install openssl openssl-devel -y
# 下载并解压源码
[root@localhost ~]# tar xf Python-3.9.7.tgz
[root@localhost ~]# cd Python-3.9.7
# 编译安装
[root@localhost Python-3.9.7]# ./configure --prefix=/usr/local/python3
[root@localhost Python-3.9.7]# make && make install
# 创建软链接
[root@localhost Python-3.9.7]# ln -s /usr/local/python3/bin/python3 /usr/bin/python3
# 验证
[root@localhost ~]# python3 -V
Python 3.9.7
Windows 安装（推荐 Anaconda/Miniconda）：
- 访问 https://www.anaconda.com 下载 Miniconda
- 安装后打开 Anaconda Prompt
- 验证：
python --version
1.4 虚拟环境
想一想：如果项目 A 需要 requests 2.25，项目 B 需要 requests 2.28，怎么办？
就像不同的实验课需要不同的试剂，不同的项目也需要不同的 Python 环境。虚拟环境就是为每个项目创建独立的"实验室"。
# 创建虚拟环境
python3 -m venv myproject_env
# 激活虚拟环境（Linux）
source myproject_env/bin/activate
# 激活虚拟环境（Windows）
myproject_env\Scripts\activate
# 安装项目专属包
pip install requests==2.28.0
# 退出虚拟环境
deactivate
2. 基本使用：交互式与脚本模式
2.1 交互式模式——即问即答
类比：交互式模式就像在计算器上按一个算一个，马上就能看到结果。
>>> print("Hello, World!")
Hello, World!
>>> 2 + 3
5
>>> 100 / 3
33.333333333333336
>>> exit()
提示：安装
ipython（pip install ipython）可以获得更好的交互体验，支持代码补全和语法高亮。
2.2 脚本模式——编写程序
类比：脚本模式就像把解题步骤写在纸上，一次性执行。
创建文件 hello.py：
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
print("Hello, World!")
print("欢迎来到 Python 的世界！")
运行方式：
# Linux
[root@localhost ~]# python3 hello.py
Hello, World!
欢迎来到 Python 的世界！
# Windows
C:\> python hello.py
2.3 文件类型
| 扩展名 | 说明 |
|---|---|
.py | Python 源代码文件 |
.pyc | 编译后的字节码文件（加速加载） |
.pyo | 优化后的字节码文件 |
思考题：为什么运行一次 .py 文件后会出现 __pycache__ 文件夹？
答案：Python 解释器会自动将 .py 文件编译为 .pyc 存入该目录，下次运行时若源码未变则跳过编译，直接加载 .pyc，从而加快启动速度。
2.4 编码声明
在 Python 3 中，源码默认使用 UTF-8 编码，一般不需要额外声明。但如果文件包含中文，建议在文件开头加上：
# -*- coding: utf-8 -*-
3. 变量与数据类型
3.1 变量——带标签的盒子
类比：变量就像一个贴了标签的盒子，你可以往盒子里放东西（赋值），也可以换掉里面的东西。盒子上的标签就是变量名。
>>> name = "张三" # 字符串
>>> age = 17 # 整数
>>> height = 1.75 # 浮点数
>>> is_student = True # 布尔值
>>> print(f"{name}，{age}岁，身高{height}米")
张三，17岁，身高1.75米
变量命名规则：
- 由字母、数字、下划线组成
- 不能以数字开头
- 不能使用 Python 关键字（如
if、for、class等） - 建议使用有意义的英文命名，如
user_name而非a
>>> # 正确的命名
>>> user_name = "admin"
>>> server_count = 5
>>> _private_var = "secret"
>>> # 错误的命名
>>> 1user = "admin" # SyntaxError: 数字开头
>>> class = "math" # SyntaxError: 关键字
3.2 Python 的四种基本数据类型
| 类型 | 说明 | 示例 |
|---|---|---|
int | 整数 | 42, -100, 0 |
float | 浮点数（小数） | 3.14, -0.5, 2.0 |
str | 字符串 | "hello", 'world' |
bool | 布尔值 | True, False |
>>> a = 100
>>> type(a)
<class 'int'>
>>> b = 3.14
>>> type(b)
<class 'float'>
>>> c = "hello"
>>> type(c)
<class 'str'>
>>> d = True
>>> type(d)
<class 'bool'>
3.3 类型转换
运维场景：从配置文件读取的端口号是字符串 "8080"，需要转为整数才能进行比较。
>>> port_str = "8080"
>>> port_int = int(port_str) # 字符串转整数
>>> port_int
8080
>>> price = 9.99
>>> price_str = str(price) # 浮点数转字符串
>>> price_str
'9.99'
>>> # 常见转换函数
>>> int("123") # 123
>>> float("3.14") # 3.14
>>> str(100) # '100'
>>> bool(0) # False
>>> bool("") # False
>>> bool("hello") # True（非空字符串为 True）
3.4 多重赋值与变量交换
# 多重赋值
>>> x = y = z = 0
# 多元赋值
>>> a, b, c = 1, 2, 3
# 变量交换（Python 的优雅写法）
>>> x, y = 10, 20
>>> x, y = y, x # 一行完成交换！
>>> print(x, y)
20 10
想一想：在 C 语言中交换两个变量需要一个临时变量
temp，Python 为什么不需要？因为 Python 在等号右边先创建了一个元组
(y, x)，然后再解包赋值给左边的x, y。
3.5 变量的引用机制
重要概念：Python 中，变量不是"盒子"，而是"标签"。同一个值可以贴多个标签。
>>> a = 123
>>> b = 123
>>> id(a) # 查看内存地址
140712345678912
>>> id(b) # 地址相同！
140712345678912
>>> a = 456 # a 换了个标签，123 还在内存中
>>> id(a) # 地址变了
140712345679008
4. 运算符
4.1 算术运算符
类比：算术运算符就是数学课上的加减乘除，只不过多了几个"新朋友"。
>>> 10 + 3 # 加法
13
>>> 10 - 3 # 减法
7
>>> 10 * 3 # 乘法
30
>>> 10 / 3 # 除法（Python3 返回浮点数）
3.3333333333333335
>>> 10 // 3 # 整除（只取整数部分）
3
>>> 10 % 3 # 取余（求模）
1
>>> 2 ** 10 # 幂运算（2的10次方）
1024
运维实用示例：
# 计算磁盘使用率
total_disk = 500 # GB
used_disk = 347 # GB
usage_percent = (used_disk / total_disk) * 100
print(f"磁盘使用率: {usage_percent:.1f}%")
# 输出: 磁盘使用率: 69.4%
# 判断端口号是否合法（1-65535）
port = 8080
is_valid = 1 <= port <= 65535
print(f"端口 {port} 是否合法: {is_valid}")
4.2 比较运算符
>>> 5 > 3 # True
>>> 5 == 3 # False
>>> 5 != 3 # True
>>> 5 >= 5 # True
>>> "abc" == "abc" # True（字符串比较）
4.3 逻辑运算符
类比：逻辑运算符就像做条件判断——"与"是两票都赞成才通过，"或"是一票赞成就可以，"非"是唱反调。
>>> True and False # 与：两个都为 True 才是 True
False
>>> True or False # 或：一个为 True 就是 True
True
>>> not True # 非：取反
False
# 实际例子：判断服务器是否正常
cpu_ok = True
mem_ok = True
disk_ok = False
server_healthy = cpu_ok and mem_ok and disk_ok
print(f"服务器状态正常: {server_healthy}")
# 输出: 服务器状态正常: False
4.4 赋值运算符
>>> x = 10
>>> x += 5 # 等价于 x = x + 5
>>> x
15
>>> x -= 3 # 等价于 x = x - 3
>>> x
12
>>> x *= 2 # 等价于 x = x * 2
>>> x
24
>>> x //= 5 # 等价于 x = x // 5
>>> x
4
思考题：以下代码输出什么？为什么？
a = 2 + 5 * 3
print(a)
答案：17。因为乘法优先级高于加法，先算 5*3=15，再算 2+15=17。
5. 流程控制：if/elif/else
5.1 为什么需要流程控制？
类比：流程控制就像导航软件——根据路况（条件）选择不同的路线（代码分支）。
问题：如何编写一个脚本，根据磁盘使用率发出不同的警告？
5.2 基本语法
# 单分支
if 条件:
执行代码
# 双分支
if 条件:
执行代码A
else:
执行代码B
# 多分支
if 条件1:
执行代码A
elif 条件2:
执行代码B
elif 条件3:
执行代码C
else:
执行代码D
5.3 实战：磁盘使用率监控
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
usage = int(input("请输入磁盘使用率(0-100): "))
if usage >= 90:
print("【严重】磁盘使用率超过90%，请立即清理！")
elif usage >= 80:
print("【警告】磁盘使用率超过80%，建议清理。")
elif usage >= 70:
print("【提醒】磁盘使用率超过70%，请关注。")
else:
print("【正常】磁盘状态良好。")
运行示例：
请输入磁盘使用率(0-100): 85
【警告】磁盘使用率超过80%，建议清理。
5.4 嵌套条件
#!/usr/bin/env python3
is_raining = int(input("下雨吗？[1=是, 0=否]: "))
if is_raining == 1:
has_umbrella = int(input("有伞吗？[1=有, 0=没有]: "))
if has_umbrella == 1:
print("带伞出门")
else:
print("在家待着吧！")
else:
print("直接出门")
5.5 三元运算（简洁写法）
>>> age = 20
>>> status = "成年" if age >= 18 else "未成年"
>>> print(status)
成年
想一想：以下成绩评级代码有什么问题？
score = 95
if score >= 90:
print("A")
if score >= 80:
print("B")
if score >= 70:
print("C")
else:
print("不及格")
问题：输出 A B C，因为每个 if 是独立的！应该用 elif 使它们互斥。
6. for 循环
6.1 为什么需要循环？
类比：for 循环就像购物清单——你逐项完成，直到清单上的东西全部买完。
问题：如何快速检查 100 台服务器的连通性？逐台手动 ping 显然不现实。
6.2 基本语法
for 变量 in 序列:
循环体
6.3 range() 函数——数字序列生成器
# range(stop) —— 从0到stop-1
>>> for i in range(5):
... print(i, end=" ")
0 1 2 3 4
# range(start, stop) —— 从start到stop-1
>>> for i in range(1, 6):
... print(i, end=" ")
1 2 3 4 5
# range(start, stop, step) —— 指定步长
>>> for i in range(0, 10, 2):
... print(i, end=" ")
0 2 4 6 8
6.4 遍历序列
# 遍历字符串
for char in "Hello":
print(char, end=" ")
# 输出: H e l l o
# 遍历列表
servers = ["web01", "web02", "db01", "cache01"]
for server in servers:
print(f"正在检查服务器: {server}")
# 使用 enumerate() 同时获取索引和值
for index, server in enumerate(servers):
print(f"第{index+1}台服务器: {server}")
输出：
第1台服务器: web01
第2台服务器: web02
第3台服务器: db01
第4台服务器: cache01
6.5 实战：计算 1 到 100 的和
total = 0
for i in range(1, 101):
total += i
print(f"1到100的和 = {total}")
# 输出: 1到100的和 = 5050
6.6 实战：九九乘法表
for i in range(1, 10):
for j in range(1, i + 1):
print(f"{j}x{i}={i*j}", end="\t")
print() # 换行
输出：
1x1=1
1x2=2 2x2=4
1x3=3 2x3=6 3x3=9
1x4=4 2x4=8 3x4=12 4x4=16
...
7. while 循环
7.1 while vs for
类比：for 循环像按计划执行的旅行（已知天数），while 循环像条件触发的警报——只要条件满足就一直响。
7.2 基本语法
while 条件:
循环体
7.3 基本示例
# 倒计时
count = 5
while count > 0:
print(f"倒计时: {count}")
count -= 1
print("发射！")
7.4 break 和 continue
# break：跳出整个循环
while True:
user_input = input("输入命令 (q退出): ")
if user_input == 'q':
print("退出程序")
break
print(f"执行命令: {user_input}")
# continue：跳过本次循环，继续下一次
for i in range(10):
if i == 5:
continue # 跳过5
print(i, end=" ")
# 输出: 0 1 2 3 4 6 7 8 9
7.5 while-else 语法
想一想：怎么知道循环是正常结束的还是被 break 中断的？
# while-else: 循环正常结束时执行 else
count = 0
while count < 5:
print(count, end=" ")
count += 1
else:
print("\n循环正常结束！")
# 输出: 0 1 2 3 4
# 循环正常结束！
# 如果被 break 中断，else 不执行
count = 0
while count < 5:
if count == 3:
print("\n被中断了！")
break
print(count, end=" ")
count += 1
else:
print("这行不会执行")
# 输出: 0 1 2
# 被中断了！
7.6 实战：模拟登录（3 次机会）
#!/usr/bin/env python3
import getpass
correct_user = "admin"
correct_pass = "123456"
max_attempts = 3
for attempt in range(1, max_attempts + 1):
username = input(f"第{attempt}次登录 - 用户名: ")
password = input("密码: ")
if username == correct_user and password == correct_pass:
print("登录成功！欢迎回来，管理员。")
break
else:
remaining = max_attempts - attempt
if remaining > 0:
print(f"用户名或密码错误，还剩{remaining}次机会。\n")
else:
print("登录失败次数过多，账号已锁定，请明天再试。")
8. 循环控制进阶
8.1 for-else 语法
# 查找素数的经典写法
for n in range(2, 20):
for x in range(2, n):
if n % x == 0:
break
else:
# 循环正常结束（没有 break），说明是素数
print(f"{n} 是素数")
输出：
2 是素数
3 是素数
5 是素数
7 是素数
11 是素数
13 是素数
17 是素数
19 是素数
8.2 嵌套循环
# 打印矩形
for i in range(4): # 4行
for j in range(8): # 每行8个星号
print("*", end="")
print() # 换行
输出：
********
********
********
********
8.3 循环优化技巧
# 不好的写法：在循环内重复计算
for i in range(1000):
result = len(some_list) * 2 + i # len() 每次都调用
# 好的写法：提前计算
list_len = len(some_list)
double_len = list_len * 2
for i in range(1000):
result = double_len + i
思考题：编写脚本，生成 1000 个文件名（file_0001.txt ~ file_1000.txt），如何实现？
for i in range(1, 1001):
filename = f"file_{i:04d}.txt"
# 创建文件...
print(filename)
# 输出: file_0001.txt, file_0002.txt, ..., file_1000.txt
9. 数据类型深入
9.1 列表 (list)——购物清单
类比：列表就像一张购物清单，可以添加、删除、修改项目，顺序很重要。
# 创建列表
servers = ["web01", "web02", "db01", "cache01"]
ports = [80, 443, 3306, 6379]
# 访问元素（索引从0开始）
>>> servers[0]
'web01'
>>> servers[-1] # 最后一个元素
'cache01'
# 切片
>>> servers[1:3] # 取索引1到2
['web02', 'db01']
>>> servers[:2] # 前两个
['web01', 'web02']
# 常用方法
servers.append("monitor01") # 末尾添加
servers.insert(0, "gateway") # 指定位置插入
servers.remove("web02") # 删除指定元素
popped = servers.pop() # 弹出最后一个元素
servers.sort() # 排序
print(len(servers)) # 长度
# 列表推导式（Pythonic 写法）
squares = [x**2 for x in range(1, 6)]
print(squares) # [1, 4, 9, 16, 25]
# 过滤：找出大于80的端口
big_ports = [p for p in ports if p > 1000]
print(big_ports) # [3306, 6379]
9.2 元组 (tuple)——不可修改的列表
类比：元组就像一个密封的档案袋——内容一旦放入就不能更改，适合存放不应该被修改的数据。
# 创建元组
server_info = ("web01", "192.168.1.10", 80)
coordinates = (39.9, 116.4)
# 访问
>>> server_info[0]
'web01'
# 不能修改！
>>> server_info[0] = "web02"
TypeError: 'tuple' object does not support item assignment
# 元组解包
hostname, ip, port = server_info
print(f"主机: {hostname}, IP: {ip}, 端口: {port}")
# 单元素元组要加逗号
single = (42,) # 这是元组
not_tuple = (42) # 这只是数字 42
9.3 字典 (dict)——带标签的收纳柜
类比：字典就像一个带标签的收纳柜，每个格子上贴着标签（键），里面放着物品（值），通过标签快速找到对应的物品。
# 创建字典
server = {
"hostname": "web01",
"ip": "192.168.1.10",
"os": "CentOS 7.9",
"cpu_cores": 4,
"memory_gb": 8
}
# 访问
>>> server["hostname"]
'web01'
>>> server.get("disk", "未配置") # 安全获取，不存在返回默认值
'未配置'
# 增删改
server["disk_gb"] = 500 # 添加
server["memory_gb"] = 16 # 修改
del server["os"] # 删除
# 遍历
for key, value in server.items():
print(f"{key}: {value}")
# 实用：统计字符出现次数
text = "hello world"
char_count = {}
for char in text:
char_count[char] = char_count.get(char, 0) + 1
print(char_count)
# {'h': 1, 'e': 1, 'l': 3, 'o': 2, ' ': 1, 'w': 1, 'r': 1, 'd': 1}
9.4 集合 (set)——去重利器
类比：集合就像一个签到表——每个人只能签到一次，自动去重。
# 创建集合
ips = {"192.168.1.1", "192.168.1.2", "192.168.1.1"} # 自动去重
print(ips) # {'192.168.1.1', '192.168.1.2'}
# 集合运算（数学老师会喜欢这个！）
group_a = {"张三", "李四", "王五"}
group_b = {"王五", "赵六", "钱七"}
print(group_a & group_b) # 交集: {'王五'}
print(group_a | group_b) # 并集: {'张三', '李四', '王五', '赵六', '钱七'}
print(group_a - group_b) # 差集: {'张三', '李四'}
# 实用：日志去重
log_ips = ["10.0.0.1", "10.0.0.2", "10.0.0.1", "10.0.0.3", "10.0.0.2"]
unique_ips = set(log_ips)
print(f"共 {len(unique_ips)} 个独立IP")
# 输出: 共 3 个独立IP
9.5 字符串方法精选
text = " Hello, World! "
# 常用方法
text.strip() # 去除两端空白 -> "Hello, World!"
text.lower() # 转小写 -> " hello, world! "
text.upper() # 转大写 -> " HELLO, WORLD! "
text.replace("World", "Python") # 替换
text.split(",") # 按逗号分割 -> [' Hello', ' World! ']
text.find("World") # 查找位置 -> 9（找不到返回-1）
text.count("l") # 计数 -> 3
text.startswith(" H") # True
text.endswith("! ") # True
# 格式化（f-string，推荐）
hostname = "web01"
cpu = 75.5
print(f"服务器 {hostname} CPU使用率: {cpu:.1f}%")
# join：用指定字符连接列表
path_parts = ["/usr", "local", "bin"]
full_path = "/".join(path_parts)
print(full_path) # /usr/local/bin
10. 函数
10.1 为什么需要函数？
类比：函数就像菜谱卡片——把做菜步骤写在卡片上，每次需要做这道菜时，拿出卡片照着做就行，不用重新想。
问题：如果多个地方需要检查服务器端口是否合法，每次都复制粘贴代码？
10.2 定义与调用
def check_port(port):
"""检查端口号是否合法（1-65535）"""
if 1 <= port <= 65535:
return True
return False
# 调用
print(check_port(80)) # True
print(check_port(70000)) # False
print(check_port(443)) # True
10.3 参数详解
# 1. 位置参数
def greet(name, greeting):
print(f"{greeting}, {name}!")
greet("张老师", "您好") # 您好, 张老师!
# 2. 默认参数
def connect_server(host, port=22, timeout=30):
print(f"连接 {host}:{port}，超时{timeout}秒")
connect_server("192.168.1.10") # 使用默认端口22
connect_server("192.168.1.10", port=8080) # 指定端口
# 3. *args：接收任意数量的位置参数（打包为元组）
def sum_all(*args):
print(f"收到参数: {args}")
return sum(args)
print(sum_all(1, 2, 3)) # 6
print(sum_all(10, 20, 30, 40)) # 100
# 4. **kwargs：接收任意数量的关键字参数（打包为字典）
def config_server(**kwargs):
for key, value in kwargs.items():
print(f" {key} = {value}")
print("服务器配置:")
config_server(host="192.168.1.10", port=80, workers=4)
输出：
服务器配置:
host = 192.168.1.10
port = 80
workers = 4
10.4 返回值
# 返回多个值（实际返回的是元组）
def get_server_status():
cpu = 45.2
memory = 62.8
disk = 78.5
return cpu, memory, disk
cpu, mem, disk = get_server_status()
print(f"CPU: {cpu}%, 内存: {mem}%, 磁盘: {disk}%")
# 没有 return 的函数返回 None
def say_hello():
print("Hello!")
result = say_hello()
print(result) # None
10.5 lambda 匿名函数
类比：lambda 就像一次性便签——写个简短的公式，用完就扔。
# 普通函数
def square(x):
return x ** 2
# lambda 等价写法
square = lambda x: x ** 2
# 常见用法：配合 sorted、map、filter
servers = [
{"name": "web01", "cpu": 75},
{"name": "db01", "cpu": 90},
{"name": "cache01", "cpu": 30},
]
# 按 CPU 使用率排序
sorted_servers = sorted(servers, key=lambda s: s["cpu"])
for s in sorted_servers:
print(f"{s['name']}: CPU {s['cpu']}%")
# 输出:
# cache01: CPU 30%
# web01: CPU 75%
# db01: CPU 90%
10.6 变量作用域
global_var = "我是全局变量"
def my_function():
local_var = "我是局部变量"
print(global_var) # 可以访问全局变量
print(local_var) # 可以访问局部变量
my_function()
print(global_var) # OK
# print(local_var) # NameError: 局部变量在函数外不可见
# 使用 global 关键字在函数内修改全局变量
counter = 0
def increment():
global counter
counter += 1
increment()
increment()
print(counter) # 2
11. switch 语句替代方案
11.1 问题：Python 没有 switch 语句
很多编程语言都有 switch/case 语句来实现多分支选择，但 Python 没有提供这个语法。
想一想：如果要实现一个简单计算器，用 if/elif 写起来是不是很冗长？
11.2 解决方案：字典派发表
核心思路：利用字典的键值对来模拟 switch 的功能。
# 定义运算函数
def add(x, y):
return x + y
def subtract(x, y):
return x - y
def multiply(x, y):
return x * y
def divide(x, y):
if y == 0:
return "错误：除数不能为0"
return x / y
# 用字典模拟 switch
operations = {
"+": add,
"-": subtract,
"*": multiply,
"/": divide
}
def calculator(x, operator, y):
# get() 获取对应函数，找不到返回 None
func = operations.get(operator)
if func:
return func(x, y)
return "不支持的运算符"
# 测试
print(calculator(10, "+", 5)) # 15
print(calculator(10, "-", 3)) # 7
print(calculator(6, "*", 7)) # 42
print(calculator(15, "/", 4)) # 3.75
print(calculator(10, "%", 3)) # 不支持的运算符
11.3 实战：命令行菜单系统
#!/usr/bin/env python3
def show_status():
print("=== 系统状态 ===")
print("CPU: 45% 内存: 62% 磁盘: 78%")
def show_logs():
print("=== 最近日志 ===")
print("[INFO] 系统正常运行中...")
def backup_data():
print("=== 数据备份 ===")
print("备份完成！")
def exit_program():
print("再见！")
exit()
# 菜单字典
menu = {
"1": ("查看系统状态", show_status),
"2": ("查看日志", show_logs),
"3": ("数据备份", backup_data),
"0": ("退出", exit_program),
}
while True:
print("\n===== 运维菜单 =====")
for key, (desc, _) in menu.items():
print(f" ({key}) {desc}")
choice = input("请选择操作: ").strip()
action = menu.get(choice)
if action:
desc, func = action
func()
else:
print("无效选项，请重新输入。")
12. 模块与代码封装
12.1 什么是模块？
类比：模块就像工具箱里的不同工具——螺丝刀处理螺丝，扳手处理螺母。把功能相关的代码放在不同文件中，各司其职。
12.2 导入方式
# 方式1：导入整个模块
import os
print(os.getcwd()) # 获取当前目录
# 方式2：导入特定函数
from os.path import exists
print(exists("/etc/passwd")) # True
# 方式3：起别名
import subprocess as sp
result = sp.run(["ls", "-l"], capture_output=True, text=True)
# 方式4：导入所有（不推荐，容易命名冲突）
from os import *
12.3 自定义模块
创建文件 server_utils.py：
# server_utils.py —— 服务器工具模块
def check_disk_usage(threshold=80):
"""检查磁盘使用率"""
import shutil
total, used, free = shutil.disk_usage("/")
usage = (used / total) * 100
if usage > threshold:
return f"警告：磁盘使用率 {usage:.1f}% 超过阈值 {threshold}%"
return f"正常：磁盘使用率 {usage:.1f}%"
def format_bytes(size_bytes):
"""将字节数格式化为可读大小"""
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if size_bytes < 1024:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024
if __name__ == "__main__":
# 只有直接运行此文件时才执行，被导入时不执行
print(check_disk_usage())
print(format_bytes(1073741824))
在另一个文件中使用：
# main.py
import server_utils
print(server_utils.check_disk_usage(90))
print(server_utils.format_bytes(5242880))
12.4 __name__ 的妙用
想一想：为什么上面的代码要用 if __name__ == "__main__"？
当一个 .py 文件被直接运行时，__name__ 的值为 "__main__"；当它被作为模块导入时，__name__ 的值为模块名。这个机制让我们可以在模块中写测试代码，而不用担心导入时自动执行。
# 模块被导入时 __name__ = 模块名
# 模块被直接运行时 __name__ = "__main__"
if __name__ == "__main__":
print("这是直接运行")
else:
print(f"这是被 {__name__} 导入的")
12.5 常用运维模块
| 模块 | 用途 |
|---|---|
os | 操作系统接口（路径、文件、目录） |
sys | 系统参数（命令行参数、路径） |
subprocess | 执行系统命令 |
shutil | 高级文件操作（复制、移动、删除） |
json | JSON 数据处理 |
datetime | 日期时间处理 |
re | 正则表达式 |
logging | 日志记录 |
socket | 网络编程 |
13. 反射机制
13.1 什么是反射？
类比：反射就像你不用提前知道餐厅有什么菜——你到了餐厅，看菜单（dir()），问服务员有没有某道菜（hasattr()），然后点菜（getattr()）。
核心思想：通过字符串来动态访问对象（模块）的属性和方法。
13.2 四大反射函数
# 假设有一个模块 commons.py
# commons.py:
def login():
print("登录页面")
def home():
print("网站主页")
def logout():
print("登出页面")
问题：用户输入不同的 URL，如何调用对应的函数？难道写 100 个 elif？
import commons
def run():
url = input("输入页面URL: ").strip()
# hasattr：检查模块中是否有这个成员
if hasattr(commons, url):
# getattr：动态获取成员
func = getattr(commons, url)
func() # 调用函数
else:
print("404 页面不存在")
run()
13.3 反射函数一览
| 函数 | 作用 | 示例 |
|---|---|---|
hasattr(obj, name) | 判断对象是否有某属性 | hasattr(commons, "login") |
getattr(obj, name) | 获取对象的属性/方法 | getattr(commons, "login") |
setattr(obj, name, val) | 设置对象的属性 | setattr(obj, "name", "value") |
delattr(obj, name) | 删除对象的属性 | delattr(obj, "name") |
13.4 动态导入模块
# __import__() 可以根据字符串动态导入模块
module_name = "os.path"
mod = __import__(module_name)
# 实用场景：根据配置文件动态加载不同的处理模块
handlers = ["handler_login", "handler_dashboard", "handler_report"]
for handler_name in handlers:
module = __import__(handler_name)
# 调用模块中的 process 函数
if hasattr(module, "process"):
getattr(module, "process")()
思考题：反射在 Web 框架中有什么应用场景？
答：URL 路由！根据用户访问的 URL 字符串，动态找到并调用对应的处理函数，这就是反射最典型的运维/Web 应用。
14. 正则表达式
14.1 为什么需要正则表达式？
类比：正则表达式就像搜索引擎的高级搜索——你不是找一个具体的词，而是找符合某种模式的文本。比如"找出所有手机号"、"找出所有邮箱地址"。
14.2 re 模块基础
import re
text = "服务器IP: 192.168.1.100, 备用IP: 10.0.0.1, 端口: 8080"
# findall：找出所有匹配的内容
ips = re.findall(r'\d+\.\d+\.\d+\.\d+', text)
print(ips) # ['192.168.1.100', '10.0.0.1']
# search：找到第一个匹配
match = re.search(r'端口: (\d+)', text)
if match:
print(f"端口号: {match.group(1)}") # 端口号: 8080
# match：从字符串开头匹配
result = re.match(r'服务器', text)
print(result) # <re.Match object>（匹配成功）
# sub：替换
cleaned = re.sub(r'\d+', 'X', text)
print(cleaned)
# 服务器IP: X.X.X.X, 备用IP: X.X.X, 端口: X
14.3 常用正则符号
| 符号 | 含义 | 示例 |
|---|---|---|
. | 匹配任意字符（除换行） | a.b 匹配 acb, a3b |
\d | 匹配数字 [0-9] | \d{3} 匹配 123 |
\D | 匹配非数字 | \D+ 匹配 abc |
\w | 匹配字母数字下划线 | \w+ 匹配 hello_123 |
\s | 匹配空白字符 | \s+ 匹配空格、Tab |
^ | 匹配行首 | ^hello |
$ | 匹配行尾 | world$ |
* | 前一个字符出现0次或多次 | ab* 匹配 a, ab, abb |
+ | 前一个字符出现1次或多次 | ab+ 匹配 ab, abb |
? | 前一个字符出现0次或1次 | colou?r 匹配 color, colour |
{m,n} | 前一个字符出现m到n次 | \d{2,4} 匹配 12, 123, 1234 |
[] | 字符集 | [abc] 匹配 a、b或c |
[^] | 取反 | [^0-9] 匹配非数字 |
14.4 实战示例
import re
# 1. 提取日志中的 IP 地址
log_line = '192.168.1.50 - - [01/Jan/2024:10:30:00] "GET /index.html HTTP/1.1" 200'
ip = re.match(r'^(\d+\.\d+\.\d+\.\d+)', log_line)
if ip:
print(f"访问IP: {ip.group(1)}")
# 输出: 访问IP: 192.168.1.50
# 2. 验证手机号
def is_phone(number):
pattern = r'^1[3-9]\d{9}$'
return bool(re.match(pattern, number))
print(is_phone("13812345678")) # True
print(is_phone("12345")) # False
# 3. 提取邮箱
text = "联系我: admin@example.com 或 support@test.org"
emails = re.findall(r'[\w.]+@[\w.]+\.\w+', text)
print(emails) # ['admin@example.com', 'support@test.org']
# 4. 解析 /etc/passwd 格式
passwd_line = "root:x:0:0:root:/root:/bin/bash"
parts = re.split(r':', passwd_line)
print(f"用户: {parts[0]}, UID: {parts[2]}, Shell: {parts[6]}")
# 输出: 用户: root, UID: 0, Shell: /bin/bash
# 5. 编译正则（提升性能，适合重复使用）
ip_pattern = re.compile(r'\d+\.\d+\.\d+\.\d+')
results = ip_pattern.findall("server1: 10.0.0.1, server2: 10.0.0.2")
print(results) # ['10.0.0.1', '10.0.0.2']
思考题：如何匹配 IP 地址 192.168.1.1 到 192.168.1.255 这个网段的所有 IP？
# 简单版（不严格验证范围）
pattern = r'192\.168\.1\.\d{1,3}'
# 严格版（验证 1-255）
pattern = r'192\.168\.1\.([1-9]|[1-9]\d|1\d{2}|2[0-4]\d|25[0-5])'
15. 文件 IO 与操作系统交互
15.1 文件读写基础
类比：操作文件就像使用笔记本——先打开（open），然后读或写（read/write），最后合上（close）。
15.2 文件打开模式
| 模式 | 说明 |
|---|---|
r | 只读（默认） |
w | 写入（会清空原文件！） |
a | 追加（在文件末尾添加） |
r+ | 读写 |
w+ | 读写（会清空原文件） |
a+ | 读写（追加模式） |
b | 二进制模式（配合其他模式使用，如 rb） |
15.3 使用 with 语句（推荐）
问题：如果读取文件时出错了，close() 就不会被执行，文件会一直占用着。
# 不推荐：手动关闭
fo = open("test.txt", "r")
content = fo.read()
fo.close() # 如果上面出错，这行不会执行
# 推荐：with 语句自动关闭
with open("test.txt", "r", encoding="utf-8") as fo:
content = fo.read()
print(content)
# 离开 with 块后自动关闭文件
15.4 文件读写方法
# 读取整个文件
with open("config.txt", "r", encoding="utf-8") as f:
content = f.read() # 读取全部内容（字符串）
# 逐行读取
with open("config.txt", "r", encoding="utf-8") as f:
for line in f: # 最 Pythonic 的方式
print(line.strip())
# 读取所有行到列表
with open("config.txt", "r", encoding="utf-8") as f:
lines = f.readlines() # 返回列表
for line in lines:
print(line.strip())
# 写入文件
with open("output.txt", "w", encoding="utf-8") as f:
f.write("第一行\n")
f.write("第二行\n")
# 写入多行
with open("output.txt", "w", encoding="utf-8") as f:
lines = ["第一行\n", "第二行\n", "第三行\n"]
f.writelines(lines)
# 追加内容
with open("log.txt", "a", encoding="utf-8") as f:
f.write("[2024-01-01 10:00:00] 系统启动\n")
15.5 os 模块——操作系统接口
import os
# 当前工作目录
print(os.getcwd()) # /home/user
# 切换目录
os.chdir("/tmp")
# 创建目录
os.mkdir("test_dir") # 创建单级目录
os.makedirs("a/b/c", exist_ok=True) # 递归创建目录
# 列出目录内容
files = os.listdir(".")
print(files)
# 判断文件/目录
print(os.path.isfile("test.txt")) # True/False
print(os.path.isdir("test_dir")) # True/False
print(os.path.exists("/etc/passwd")) # True
# 路径操作
path = "/home/user/documents/report.txt"
print(os.path.dirname(path)) # /home/user/documents
print(os.path.basename(path)) # report.txt
print(os.path.join("/home", "user", "test.py")) # /home/user/test.py
print(os.path.splitext("report.txt")) # ('report', '.txt')
# 删除文件和目录
os.remove("test.txt") # 删除文件
os.rmdir("empty_dir") # 删除空目录
# 环境变量
print(os.environ.get("HOME")) # /home/user
15.6 shutil 模块——高级文件操作
import shutil
# 复制文件
shutil.copy("source.txt", "backup.txt") # 复制文件
shutil.copy2("source.txt", "backup2.txt") # 复制文件（保留元数据）
shutil.copytree("src_dir", "dst_dir") # 复制整个目录
# 移动/重命名
shutil.move("old_name.txt", "new_name.txt")
# 删除目录树
shutil.rmtree("dir_to_delete")
# 获取磁盘使用情况
total, used, free = shutil.disk_usage("/")
print(f"总计: {total // (1024**3)} GB")
print(f"已用: {used // (1024**3)} GB")
print(f"可用: {free // (1024**3)} GB")
15.7 实战：日志文件分析器
#!/usr/bin/env python3
import re
from collections import Counter
def analyze_log(log_file):
"""分析 Apache/Nginx 访问日志"""
ip_counter = Counter()
status_counter = Counter()
with open(log_file, "r", encoding="utf-8") as f:
for line in f:
# 提取IP地址（假设在行首）
ip_match = re.match(r'^(\d+\.\d+\.\d+\.\d+)', line)
if ip_match:
ip_counter[ip_match.group(1)] += 1
# 提取HTTP状态码
status_match = re.search(r'" (\d{3}) ', line)
if status_match:
status_counter[status_match.group(1)] += 1
print("=== IP 访问排行 TOP5 ===")
for ip, count in ip_counter.most_common(5):
print(f" {ip}: {count} 次")
print("\n=== HTTP 状态码统计 ===")
for status, count in status_counter.most_common():
print(f" {status}: {count} 次")
# 使用示例
# analyze_log("/var/log/nginx/access.log")
15.8 目录遍历
import os
# 方法1：os.walk() —— 递归遍历
for root, dirs, files in os.walk("/var/log"):
for file in files:
filepath = os.path.join(root, file)
print(filepath)
# 方法2：查找特定文件
def find_files(directory, extension):
"""在目录中查找指定扩展名的文件"""
result = []
for root, dirs, files in os.walk(directory):
for file in files:
if file.endswith(extension):
result.append(os.path.join(root, file))
return result
py_files = find_files("/home/user", ".py")
for f in py_files:
print(f)
16. 异常处理
16.1 为什么需要异常处理？
类比：异常处理就像汽车的安全气囊——正常行驶时用不到，但出事故时能保命，防止程序直接崩溃。
问题：读取一个可能不存在的配置文件，如果文件不存在程序就崩溃了，怎么办？
16.2 try/except 基本语法
try:
# 可能出错的代码
with open("config.ini", "r") as f:
config = f.read()
except FileNotFoundError:
# 文件不存在时的处理
print("配置文件不存在，使用默认配置")
config = "default_config"
except PermissionError:
# 权限不足时的处理
print("权限不足，无法读取配置文件")
16.3 捕获多种异常
try:
num = int(input("请输入一个数字: "))
result = 100 / num
print(f"结果: {result}")
except ValueError:
print("输入的不是有效数字！")
except ZeroDivisionError:
print("除数不能为0！")
except Exception as e:
print(f"未知错误: {e}")
16.4 try/except/else/finally
try:
f = open("data.txt", "r", encoding="utf-8")
content = f.read()
except FileNotFoundError:
print("文件不存在")
except Exception as e:
print(f"发生错误: {e}")
else:
# try 块没有异常时执行
print(f"成功读取 {len(content)} 个字符")
finally:
# 无论如何都会执行（通常用于资源清理）
print("清理工作完成")
if 'f' in dir() and not f.closed:
f.close()
16.5 raise：主动抛出异常
def set_port(port):
if not isinstance(port, int):
raise TypeError("端口号必须是整数")
if port < 1 or port > 65535:
raise ValueError(f"端口号 {port} 超出范围(1-65535)")
print(f"端口设置为: {port}")
# 测试
try:
set_port(8080) # OK
set_port(70000) # 抛出 ValueError
except (TypeError, ValueError) as e:
print(f"设置失败: {e}")
16.6 自定义异常
class ServerError(Exception):
"""自定义：服务器错误"""
pass
class DiskFullError(ServerError):
"""自定义：磁盘已满"""
pass
def write_log(message):
import shutil
_, _, free = shutil.disk_usage("/")
free_mb = free // (1024 * 1024)
if free_mb < 100:
raise DiskFullError(f"磁盘剩余空间不足: {free_mb}MB")
# 写入日志...
print(f"日志写入: {message}")
try:
write_log("系统启动")
except DiskFullError as e:
print(f"写入失败: {e}")
except ServerError as e:
print(f"服务器错误: {e}")
16.7 常见异常类型
| 异常 | 触发场景 |
|---|---|
FileNotFoundError | 文件不存在 |
PermissionError | 权限不足 |
ValueError | 值不合法（如 int("abc")） |
TypeError | 类型错误 |
KeyError | 字典键不存在 |
IndexError | 列表索引越界 |
AttributeError | 属性不存在 |
ImportError | 模块导入失败 |
ZeroDivisionError | 除以零 |
KeyboardInterrupt | 用户按下 Ctrl+C |
17. 装饰器
17.1 从闭包说起
类比：闭包就像一个带记忆的函数——它不仅携带了代码逻辑，还"记住"了外部变量的值。
def outer():
x = 10
def inner():
print(f"x = {x}") # 引用了外部变量 x
return inner
f = outer()
f() # 输出: x = 10
闭包的两个条件：
- 函数内部定义了另一个函数
- 内部函数引用了外部函数的变量（非全局变量）
17.2 装饰器是什么？
类比：装饰器就像给手机贴膜——手机本身没变，但多了一层保护膜（新功能）。
问题：如何给多个函数添加"执行时间统计"功能，又不修改每个函数的代码？
17.3 基本装饰器
import time
# 定义装饰器
def timer(func):
def wrapper(*args, **kwargs):
start = time.time()
result = func(*args, **kwargs) # 执行原函数
end = time.time()
print(f"[{func.__name__}] 执行耗时: {end-start:.3f}秒")
return result
return wrapper
# 使用装饰器
@timer
def slow_function():
"""模拟一个耗时操作"""
time.sleep(2)
print("任务完成")
@timer
def fast_function(x, y):
return x + y
# 调用
slow_function()
# 输出:
# 任务完成
# [slow_function] 执行耗时: 2.001秒
result = fast_function(10, 20)
# 输出: [fast_function] 执行耗时: 0.000秒
print(f"结果: {result}")
17.4 装饰器的执行顺序
# @timer 等价于: slow_function = timer(slow_function)
# 也就是说，decorator 接收原函数作为参数，返回一个新函数
# 带参数的装饰器
def repeat(n):
def decorator(func):
def wrapper(*args, **kwargs):
for i in range(n):
result = func(*args, **kwargs)
return result
return wrapper
return decorator
@repeat(3)
def say_hello():
print("Hello!")
say_hello()
# 输出:
# Hello!
# Hello!
# Hello!
17.5 多装饰器叠加
def deco1(func):
def wrapper(*args, **kwargs):
print("[deco1] 开始")
result = func(*args, **kwargs)
print("[deco1] 结束")
return result
return wrapper
def deco2(func):
def wrapper(*args, **kwargs):
print("[deco2] 开始")
result = func(*args, **kwargs)
print("[deco2] 结束")
return result
return wrapper
@deco1
@deco2
def my_function():
print(" 核心逻辑 ")
my_function()
输出（注意顺序）：
[deco1] 开始
[deco2] 开始
核心逻辑
[deco2] 结束
[deco1] 结束
记忆方法：装饰器从上到下"进入"，从下到上"退出"，像洋葱一样一层层包裹。
17.6 运维实战：权限检查装饰器
import functools
def require_admin(func):
"""要求管理员权限的装饰器"""
@functools.wraps(func) # 保留原函数的文档字符串
def wrapper(user, *args, **kwargs):
if user.get("role") != "admin":
print(f"权限不足：{user['name']} 不是管理员")
return None
return func(user, *args, **kwargs)
return wrapper
@require_admin
def delete_server(user, server_name):
"""删除服务器（需要管理员权限）"""
print(f"管理员 {user['name']} 删除了服务器 {server_name}")
# 测试
admin = {"name": "张老师", "role": "admin"}
student = {"name": "小明", "role": "student"}
delete_server(admin, "web01") # 成功删除
delete_server(student, "web02") # 权限不足
18. 面向对象编程
18.1 什么是面向对象？
类比：面向过程像写菜谱——一步步操作；面向对象像设计一个机器人——把数据（属性）和行为（方法）打包在一起，让机器人自己干活。
- 类 (Class)：蓝图/模板，比如"服务器的设计图"
- 对象 (Object)：根据蓝图造出来的实物，比如"web01这台具体的服务器"
18.2 定义类与创建对象
class Server:
"""服务器类"""
# 类属性（所有实例共享）
datacenter = "北京机房"
def __init__(self, hostname, ip, cpu_cores, memory_gb):
"""构造方法：创建对象时自动调用"""
# 实例属性（每个实例独有）
self.hostname = hostname
self.ip = ip
self.cpu_cores = cpu_cores
self.memory_gb = memory_gb
self.status = "running"
def show_info(self):
"""显示服务器信息"""
print(f"主机名: {self.hostname}")
print(f"IP地址: {self.ip}")
print(f"CPU核心: {self.cpu_cores}")
print(f"内存: {self.memory_gb}GB")
print(f"状态: {self.status}")
print(f"机房: {self.datacenter}")
def shutdown(self):
"""关机"""
self.status = "stopped"
print(f"{self.hostname} 已关机")
def start(self):
"""开机"""
self.status = "running"
print(f"{self.hostname} 已启动")
# 创建对象（实例化）
web01 = Server("web01", "192.168.1.10", 4, 8)
db01 = Server("db01", "192.168.1.20", 8, 32)
web01.show_info()
print("---")
db01.shutdown()
db01.start()
18.3 封装——保护数据
class BankAccount:
def __init__(self, owner, balance):
self.owner = owner # 公有属性
self.__balance = balance # 私有属性（双下划线）
def deposit(self, amount):
"""存款"""
if amount > 0:
self.__balance += amount
print(f"存入 {amount} 元，余额: {self.__balance}")
else:
print("存款金额必须大于0")
def get_balance(self):
"""查询余额"""
return self.__balance
account = BankAccount("张老师", 10000)
account.deposit(5000) # OK
print(account.get_balance()) # 15000
# print(account.__balance) # AttributeError! 私有属性不能直接访问
18.4 继承——代码复用
class Animal:
def __init__(self, name):
self.name = name
def speak(self):
print(f"{self.name} 发出声音")
class Dog(Animal):
"""Dog 继承自 Animal"""
def speak(self):
print(f"{self.name}: 汪汪！")
class Cat(Animal):
def speak(self):
print(f"{self.name}: 喵喵！")
# 多态：同一方法，不同表现
animals = [Dog("旺财"), Cat("咪咪"), Dog("大黄")]
for animal in animals:
animal.speak()
# 输出:
# 旺财: 汪汪！
# 咪咪: 喵喵！
# 大黄: 汪汪！
18.5 运维实战：服务器管理类
class BaseServer:
"""基础服务器类"""
def __init__(self, hostname, ip):
self.hostname = hostname
self.ip = ip
self.services = []
def add_service(self, service):
self.services.append(service)
def show_status(self):
print(f"\n{'='*40}")
print(f"服务器: {self.hostname} ({self.ip})")
print(f"运行服务: {', '.join(self.services) if self.services else '无'}")
class WebServer(BaseServer):
"""Web 服务器"""
def __init__(self, hostname, ip, port=80):
super().__init__(hostname, ip) # 调用父类构造方法
self.port = port
self.add_service(f"nginx:{port}")
def deploy(self, app_name):
print(f"在 {self.hostname} 上部署应用: {app_name}")
class DatabaseServer(BaseServer):
"""数据库服务器"""
def __init__(self, hostname, ip, db_type="MySQL"):
super().__init__(hostname, ip)
self.db_type = db_type
self.add_service(f"{db_type}:3306")
def backup(self):
print(f"正在备份 {self.hostname} 的 {self.db_type} 数据...")
# 使用
web = WebServer("web01", "192.168.1.10", 8080)
db = DatabaseServer("db01", "192.168.1.20", "MySQL")
web.deploy("my_app")
web.show_status()
db.backup()
db.show_status()
输出：
在 web01 上部署应用: my_app
========================================
服务器: web01 (192.168.1.10)
运行服务: nginx:8080
正在备份 db01 的 MySQL 数据...
========================================
服务器: db01 (192.168.1.20)
运行服务: MySQL:3306
18.6 类方法与静态方法
class Config:
_instance = None
settings = {"debug": False, "port": 8080}
@classmethod
def get_setting(cls, key):
"""类方法：可以访问类属性"""
return cls.settings.get(key)
@staticmethod
def validate_port(port):
"""静态方法：不访问类属性，像独立函数"""
return 1 <= port <= 65535
# 无需实例化即可调用
print(Config.get_setting("port")) # 8080
print(Config.validate_port(8080)) # True
print(Config.validate_port(99999)) # False
19. MySQL 数据库操作
19.1 环境准备
# 安装 PyMySQL
pip install pymysql
# 安装 MySQL/MariaDB 服务端（Linux）
yum install mariadb mariadb-server -y
systemctl start mariadb
# 创建测试数据库和表
mysql -u root -p
-- MySQL 中执行
CREATE DATABASE IF NOT EXISTS testdb CHARACTER SET utf8mb4;
USE testdb;
CREATE TABLE users (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(50) NOT NULL,
passwd VARCHAR(100) NOT NULL,
money INT DEFAULT 0
);
INSERT INTO users (name, passwd, money) VALUES
('robin', '12345678', 10000),
('zorro', '87654321', 20000),
('tom', '12345678', 30000);
19.2 连接数据库
import pymysql
# 建立连接
conn = pymysql.connect(
host='127.0.0.1',
user='root',
password='123',
database='testdb',
charset='utf8mb4'
)
# 创建游标
cur = conn.cursor()
# ... 执行 SQL ...
# 关闭游标和连接
cur.close()
conn.close()
最佳实践：使用 with 语句确保资源释放：
import pymysql
def get_connection():
return pymysql.connect(
host='127.0.0.1',
user='root',
password='123',
database='testdb',
charset='utf8mb4'
)
19.3 CRUD 操作
查询 (SELECT)
import pymysql
conn = pymysql.connect(host='127.0.0.1', user='root',
password='123', database='testdb')
cur = conn.cursor()
# 查询所有数据
cur.execute("SELECT * FROM users")
all_data = cur.fetchall()
print(f"共 {len(all_data)} 条记录:")
for row in all_data:
print(f" ID={row[0]}, 姓名={row[1]}, 余额={row[3]}")
# 查询单条
cur.execute("SELECT * FROM users WHERE name='robin'")
one = cur.fetchone()
print(f"\n查询结果: {one}")
# 查询指定条数
cur.execute("SELECT * FROM users")
two_rows = cur.fetchmany(2)
print(f"\n前两条: {two_rows}")
# 返回字典格式（更易读）
cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
cur.execute("SELECT * FROM users")
for row in cur.fetchall():
print(f" {row['name']}: {row['money']} 元")
cur.close()
conn.close()
插入 (INSERT)
import pymysql
conn = pymysql.connect(host='127.0.0.1', user='root',
password='123', database='testdb')
cur = conn.cursor()
# 插入单条（使用参数化查询，防止SQL注入！）
sql = "INSERT INTO users (name, passwd, money) VALUES (%s, %s, %s)"
cur.execute(sql, ('jack', '87654321', 50000))
conn.commit() # 必须提交事务！
print(f"插入成功，新ID: {cur.lastrowid}")
# 批量插入
data = [
('lilei', '12345678', 60000),
('hanmeimei', '12345678', 70000),
]
cur.executemany(sql, data)
conn.commit()
print(f"批量插入 {cur.rowcount} 条记录")
cur.close()
conn.close()
更新 (UPDATE)
import pymysql
conn = pymysql.connect(host='127.0.0.1', user='root',
password='123', database='testdb')
cur = conn.cursor()
sql = "UPDATE users SET money = %s WHERE name = %s"
affected = cur.execute(sql, (99999, 'jack'))
conn.commit()
print(f"更新了 {affected} 条记录")
cur.close()
conn.close()
删除 (DELETE)
import pymysql
conn = pymysql.connect(host='127.0.0.1', user='root',
password='123', database='testdb')
cur = conn.cursor()
sql = "DELETE FROM users WHERE name = %s"
affected = cur.execute(sql, ('hanmeimei',))
conn.commit()
print(f"删除了 {affected} 条记录")
cur.close()
conn.close()
19.4 完整实战示例：账户管理系统
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
简单的账户管理系统
使用 MySQL 存储数据
"""
import pymysql
def get_conn():
return pymysql.connect(
host='127.0.0.1', user='root',
password='123', database='testdb',
charset='utf8mb4'
)
def register(name, passwd, money=0):
"""注册新用户"""
conn = get_conn()
cur = conn.cursor()
try:
sql = "INSERT INTO users (name, passwd, money) VALUES (%s, %s, %s)"
cur.execute(sql, (name, passwd, money))
conn.commit()
print(f"注册成功！欢迎 {name}")
except pymysql.err.IntegrityError:
print("用户名已存在！")
finally:
cur.close()
conn.close()
def login(name, passwd):
"""登录验证"""
conn = get_conn()
cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
sql = "SELECT * FROM users WHERE name=%s AND passwd=%s"
cur.execute(sql, (name, passwd))
user = cur.fetchone()
cur.close()
conn.close()
return user
def query_money(name):
"""查询余额"""
conn = get_conn()
cur = conn.cursor(cursor=pymysql.cursors.DictCursor)
cur.execute("SELECT money FROM users WHERE name=%s", (name,))
result = cur.fetchone()
cur.close()
conn.close()
if result:
print(f"{name} 的余额: {result['money']} 元")
else:
print("用户不存在")
def update_money(name, amount, operation="add"):
"""充值或消费"""
conn = get_conn()
cur = conn.cursor()
if operation == "add":
sql = "UPDATE users SET money = money + %s WHERE name = %s"
else:
sql = "UPDATE users SET money = money - %s WHERE name = %s"
cur.execute(sql, (amount, name))
conn.commit()
cur.close()
conn.close()
action = "充值" if operation == "add" else "消费"
print(f"{action} {amount} 元成功")
def main():
while True:
print("\n===== 账户管理系统 =====")
print(" (1) 注册")
print(" (2) 登录")
print(" (3) 查询余额")
print(" (4) 充值")
print(" (5) 消费")
print(" (0) 退出")
choice = input("请选择: ").strip()
if choice == "1":
name = input("用户名: ")
passwd = input("密码: ")
register(name, passwd)
elif choice == "2":
name = input("用户名: ")
passwd = input("密码: ")
user = login(name, passwd)
if user:
print(f"登录成功！余额: {user['money']} 元")
else:
print("用户名或密码错误")
elif choice == "3":
name = input("用户名: ")
query_money(name)
elif choice == "4":
name = input("用户名: ")
amount = int(input("充值金额: "))
update_money(name, amount, "add")
elif choice == "5":
name = input("用户名: ")
amount = int(input("消费金额: "))
update_money(name, amount, "sub")
elif choice == "0":
print("再见！")
break
else:
print("无效选项")
if __name__ == "__main__":
main()
安全提醒：永远使用参数化查询（
%s占位符），不要拼接 SQL 字符串，以防止 SQL 注入攻击！
附录：常用内置函数速查表
| 函数 | 说明 | 示例 |
|---|---|---|
print() | 输出 | print("hello") |
input() | 获取用户输入 | name = input("姓名: ") |
int() | 转整数 | int("123") |
float() | 转浮点数 | float("3.14") |
str() | 转字符串 | str(100) |
type() | 查看类型 | type(123) |
len() | 求长度 | len("hello") |
range() | 生成整数序列 | range(1, 10) |
enumerate() | 带索引遍历 | enumerate(["a","b"]) |
sorted() | 排序 | sorted([3,1,2]) |
reversed() | 反转 | reversed([1,2,3]) |
sum() | 求和 | sum([1,2,3]) |
max() | 最大值 | max([5,3,8]) |
min() | 最小值 | min([5,3,8]) |
abs() | 绝对值 | abs(-5) |
round() | 四舍五入 | round(3.14159, 2) |
id() | 对象内存地址 | id(x) |
isinstance() | 类型检查 | isinstance(1, int) |
open() | 打开文件 | open("f.txt", "r") |
dir() | 查看对象属性 | dir(str) |
help() | 查看帮助 | help(len) |
综合练习
练习 1：系统信息采集脚本
编写一个 Python 脚本，采集以下信息并输出为 JSON 格式：
- 主机名
- 当前时间
- 磁盘使用情况
- Python 版本
练习 2：批量文件处理
在指定目录下随机生成 100 个文件，要求：
- 文件名格式：
data_001.txt~data_100.txt - 每个文件写入随机内容（1-100行）
- 统计所有文件的总大小
练习 3：日志分析工具
读取 Nginx 访问日志，统计：
- 访问量最高的 IP（Top 10）
- 各 HTTP 状态码的数量
- 每小时的访问量分布
练习 4：猜数字游戏
程序随机生成一个 1-100 的整数
用户有 7 次机会猜
每次猜完提示"大了"或"小了"
猜中则恭喜用户并显示猜了几次
学习建议：编程就像学数学——看懂不等于会做，一定要动手敲代码。建议每学完一个章节就完成对应的练习题，遇到问题先自己查文档、看报错信息，实在解决不了再请教老师。祝学习愉快！
License:
CC BY 4.0