Python进阶教程：数据库操作入门

1
-- 建表
2
CREATE TABLE users (
3
    id INTEGER PRIMARY KEY AUTOINCREMENT,
4
    name TEXT NOT NULL,
5
    age INTEGER,
6
    email TEXT UNIQUE
7
);
8

9
-- 插入
10
INSERT INTO users (name, age, email) VALUES ("Alice", 20, "alice@example.com");
11

12
-- 查询
13
SELECT * FROM users WHERE age > 18;
14

15
-- 更新
16
UPDATE users SET age = 21 WHERE name = "Alice";
17

18
-- 删除
19
DELETE FROM users WHERE id = 1;

1
import sqlite3
2

3
# 连接（文件不存在会自动创建）
4
conn = sqlite3.connect("demo.db")
5
cursor = conn.cursor()
6

7
# 建表
8
cursor.execute("""
9
CREATE TABLE IF NOT EXISTS students (
10
    id INTEGER PRIMARY KEY AUTOINCREMENT,
11
    name TEXT NOT NULL,
12
    score REAL
13
)
14
""")
15

16
# 插入数据
17
cursor.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("张三", 88.5))
18
cursor.execute("INSERT INTO students (name, score) VALUES (?, ?)", ("李四", 92.0))
19
conn.commit()
20

21
# 查询
22
cursor.execute("SELECT * FROM students")
23
for row in cursor.fetchall():
24
    print(row)
25

26
# 关闭连接
27
conn.close()

1
import sqlite3
2

3
conn = sqlite3.connect("demo.db")
4
cursor = conn.cursor()
5

6
# 错误示范（有注入风险）
7
# name = input("输入姓名：")
8
# cursor.execute(f"SELECT * FROM students WHERE name = '{name}'")
9

10
# 正确做法：占位符 ?
11
name = "张三"
12
cursor.execute("SELECT * FROM students WHERE name = ?", (name,))
13
print(cursor.fetchall())
14

15
conn.close()

1
# pip install sqlalchemy
2
from sqlalchemy import create_engine, Column, Integer, String, Float
3
from sqlalchemy.orm import declarative_base, sessionmaker
4

5
Base = declarative_base()
6

7
class Student(Base):
8
    __tablename__ = "students_orm"
9
    id = Column(Integer, primary_key=True)
10
    name = Column(String(50), nullable=False)
11
    score = Column(Float)
12

13
# 连接数据库
14
engine = create_engine("sqlite:///orm_demo.db")
15
Base.metadata.create_all(engine)
16

17
# 增删改查
18
Session = sessionmaker(bind=engine)
19
session = Session()
20

21
student = Student(name="王五", score=95.5)
22
session.add(student)
23
session.commit()
24

25
result = session.query(Student).filter(Student.score > 90).all()
26
for s in result:
27
    print(s.name, s.score)
28

29
session.close()

1
import sqlite3
2

3
class ScoreManager:
4
    def __init__(self, db="scores.db"):
5
        self.conn = sqlite3.connect(db)
6
        self.cursor = self.conn.cursor()
7
        self.cursor.execute("""
8
            CREATE TABLE IF NOT EXISTS students (
9
                id INTEGER PRIMARY KEY AUTOINCREMENT,
10
                name TEXT NOT NULL,
11
                score REAL
12
            )
13
        """)
14
        self.conn.commit()
15

16
    def add(self, name, score):
17
        self.cursor.execute("INSERT INTO students (name, score) VALUES (?, ?)", (name, score))
18
        self.conn.commit()
19
        print(f"添加成功：{name} {score}分")
20

21
    def list_all(self):
22
        self.cursor.execute("SELECT * FROM students")
23
        for row in self.cursor.fetchall():
24
            print(row)
25

26
    def top(self, n=3):
27
        self.cursor.execute("SELECT * FROM students ORDER BY score DESC LIMIT ?", (n,))
28
        for row in self.cursor.fetchall():
29
            print(f"高分榜：{row}")
30

31
    def close(self):
32
        self.conn.close()
33

34
# 使用示例
35
mgr = ScoreManager()
36
mgr.add("张三", 88.5)
37
mgr.add("李四", 92.0)
38
mgr.add("王五", 95.5)
39
mgr.list_all()
40
mgr.top(2)
41
mgr.close()