Python 数据库与网络

一、SQL 基础回顾

sql
-- 建表CREATE TABLE users (    id INTEGER PRIMARY KEY AUTOINCREMENT,    name TEXT NOT NULL,    age INTEGER,    email TEXT UNIQUE); -- 增INSERT INTO users (name, age, email) VALUES ('Alice', 25, 'a@b.com'); -- 删DELETE FROM users WHERE name = 'Alice'; -- 改UPDATE users SET age = 26 WHERE id = 1; -- 查SELECT * FROM users WHERE age > 20;SELECT name, age FROM users ORDER BY age DESC;SELECT COUNT(*) FROM users;SELECT AVG(age) FROM users GROUP BY name HAVING COUNT(*) > 1; -- 联表SELECT u.name, o.amountFROM users uJOIN orders o ON u.id = o.user_idWHERE o.amount > 100; 

索引

sql
CREATE INDEX idx_name ON users(name);          -- 普通索引CREATE UNIQUE INDEX idx_email ON users(email);  -- 唯一索引 

二、Python 操作数据库

sqlite3（内置，无需安装）

python
import sqlite3 conn = sqlite3.connect("app.db")cursor = conn.cursor() # 建表cursor.execute("""    CREATE TABLE IF NOT EXISTS users (        id INTEGER PRIMARY KEY,        name TEXT, age INTEGER    )""") # 插入cursor.execute("INSERT INTO users (name, age) VALUES (?, ?)", ("Alice", 25))conn.commit() # 批量插入data = [("Bob", 30), ("Eve", 22)]cursor.executemany("INSERT INTO users (name, age) VALUES (?, ?)", data)conn.commit() # 查询cursor.execute("SELECT * FROM users WHERE age > ?", (20,))for row in cursor.fetchall():    print(row) # 安全：始终用参数化查询，防止 SQL 注入# ❌ cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")  --危险！# ✅ cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,)) conn.close() 

连接池（以 PyMySQL 为例）

python
from dbutils.pooled_db import PooledDBimport pymysql pool = PooledDB(    creator=pymysql,    maxconnections=10,    host="localhost", user="root", password="", database="test")conn = pool.connection()# 使用完 conn.close() 会归还到池中 

三、SQLAlchemy ORM 基础

模型定义

python
from sqlalchemy import create_engine, Column, Integer, String, ForeignKeyfrom sqlalchemy.orm import DeclarativeBase, relationship, Session class Base(DeclarativeBase):    pass class User(Base):    __tablename__ = "users"    id = Column(Integer, primary_key=True)    name = Column(String(50), nullable=False)    age = Column(Integer)    posts = relationship("Post", back_populates="author") class Post(Base):    __tablename__ = "posts"    id = Column(Integer, primary_key=True)    title = Column(String(100))    user_id = Column(Integer, ForeignKey("users.id"))    author = relationship("User", back_populates="posts") 

CRUD

python
engine = create_engine("sqlite:///blog.db")Base.metadata.create_all(engine) with Session(engine) as session:    # 增    user = User(name="Alice", age=25)    session.add(user)    session.add_all([User(name="Bob", age=30), User(name="Eve", age=22)])    session.commit()     # 查    users = session.query(User).filter(User.age > 20).all()    user = session.get(User, 1)    count = session.query(User).count()     # 排序、分页    users = session.query(User).order_by(User.age.desc()).limit(10).offset(20).all()     # 改    user = session.get(User, 1)    user.age = 26    session.commit()     # 删    session.delete(user)    session.commit() 

四、HTTP 请求（requests）

安装

bash
pip install requests 

基本用法

python
import requests # GETr = requests.get("https://api.example.com/users")r = requests.get("https://api.example.com/search", params={"q": "python"})r = requests.get("https://api.example.com/data",                 headers={"Authorization": "Bearer token123"},                 timeout=10)   # 超时（秒） # POST JSONr = requests.post("https://api.example.com/users",                  json={"name": "Alice", "age": 25}) # POST 表单r = requests.post("https://api.example.com/login",                  data={"username": "alice", "password": "secret"}) # 文件上传r = requests.post("https://api.example.com/upload",                  files={"file": open("report.pdf", "rb")}) # 响应r.status_code          # 200, 404, 500...r.json()               # 解析 JSONr.text                 # 响应文本r.content              # 响应字节r.headers              # 响应头字典r.raise_for_status()   # 非 200 时抛出异常 

Session（保持连接和 Cookie）

python
s = requests.Session()s.headers.update({"User-Agent": "MyApp/1.0"})s.auth = ("user", "pass")        # Basic Authr1 = s.get("https://api.example.com/login")  # Cookie 自动保存r2 = s.get("https://api.example.com/profile")  # 自动发送 Cookie 

异常处理

python
try:    r = requests.get("https://api.example.com/data", timeout=5)    r.raise_for_status()    data = r.json()except requests.exceptions.Timeout:    print("请求超时")except requests.exceptions.ConnectionError:    print("连接失败")except requests.exceptions.HTTPError as e:    print(f"HTTP 错误: {e}")except requests.exceptions.RequestException as e:    print(f"其他请求异常: {e}") 

五、异步编程

核心概念

• 协程（coroutine）：async def 定义的函数，调用返回协程对象
• 事件循环（event loop）：调度和执行协程
• await：挂起当前协程，等待另一个协程完成

基本用法

python
import asyncio async def fetch_data(url):    await asyncio.sleep(1)    # 模拟 IO 等待    return f"data from {url}" async def main():    # 顺序执行    r1 = await fetch_data("url1")    r2 = await fetch_data("url2")     # 并发执行    results = await asyncio.gather(        fetch_data("url1"),        fetch_data("url2"),        fetch_data("url3")    )     # 异步上下文管理器    async with aiohttp.ClientSession() as session:        async with session.get("https://api.example.com") as resp:            data = await resp.json() asyncio.run(main()) 

aiohttp 异步 HTTP

bash
pip install aiohttp 

python
import aiohttpimport asyncio async def fetch(session, url):    async with session.get(url) as resp:        return await resp.json() async def main():    async with aiohttp.ClientSession() as session:        urls = [f"https://api.example.com/item/{i}" for i in range(10)]        tasks = [fetch(session, url) for url in urls]        results = await asyncio.gather(*tasks) asyncio.run(main())