Cao Note

外观

Python 数据库操作 ​

DDL - 库管理 ​

sql
查看数据库
SHOW DATABASES;
使用数据库
USE 数据库名称;
创建数据库
CREATE DATABASE 数据库名称 [CHARSET UTF8]; [表示可选的值]
删除数据库
DROP DATABASE 数据库名称;
查看当前使用的数据库
SELECT DATABASE();
查看数据库
SHOW DATABASES;
使用数据库
USE 数据库名称;
创建数据库
CREATE DATABASE 数据库名称 [CHARSET UTF8]; [表示可选的值]
删除数据库
DROP DATABASE 数据库名称;
查看当前使用的数据库
SELECT DATABASE();

DDL - 表管理 ​

sql
查看有那些表
SHOW TABLES; //需要先选择数据库
删除表
DROP TABLE 表名称;
DROP TABLE IF EXISTS 表名称;
创建表
CREATE TABLE 表名称(
列名称 列类型,
列名称 列类型,
。。。
)
--列类型
int -- 整数
float -- 浮点数
varchar(长度) -- 文本
date -- 日期类型
timestamp -- 时间戳类型
查看有那些表
SHOW TABLES; //需要先选择数据库
删除表
DROP TABLE 表名称;
DROP TABLE IF EXISTS 表名称;
创建表
CREATE TABLE 表名称(
列名称 列类型,
列名称 列类型,
。。。
)
--列类型
int -- 整数
float -- 浮点数
varchar(长度) -- 文本
date -- 日期类型
timestamp -- 时间戳类型

DML - 数据插入 ​

语法

sql
insert into 表[(列1，列2，...)] values(值1，值2，值3，...),[(...),(...)]
insert into 表[(列1，列2，...)] values(值1，值2，值3，...),[(...),(...)]

示例

sql
-- 创建表
create table student(
id int,
name varchar(10),
age int
);
-- 仅插入id列数据
insert into student(id) values(1),(2),(3);
-- 插入全部数据
insert into student(id,name,age) values(3,'周杰伦',34),(4,'林俊杰',44)
-- 插入全部数据快捷写法
insert into student values(5,'哈哈哈',23),(6,'礼拜',33)
-- 创建表
create table student(
id int,
name varchar(10),
age int
);
-- 仅插入id列数据
insert into student(id) values(1),(2),(3);
-- 插入全部数据
insert into student(id,name,age) values(3,'周杰伦',34),(4,'林俊杰',44)
-- 插入全部数据快捷写法
insert into student values(5,'哈哈哈',23),(6,'礼拜',33)

DML - 删除 ​

基础语法

sql
delete from 表名称 [where 条件判断]
条件判断：列 操作符 值
操作符：= < > <= >= !=等等  id = 5 and name = '林俊杰'
delete from 表名称 [where 条件判断]
条件判断：列 操作符 值
操作符：= < > <= >= !=等等  id = 5 and name = '林俊杰'

示例

sql
-- 删除 id = 3 的数据
delete from student where id = 3;
-- 删除 id = 4 且 name = '林俊杰的数据'
delete from student where id = 4 and name = '林俊杰'
-- 删除 id < 3 且 name = '周杰伦的数据'
delete from student where id <= 3 and name = '周杰伦';
-- 删除全部数据
delete from student;
-- 将 id = 3 的name改成111
update student set name = '111' where id = 3;
-- 删除 id = 3 的数据
delete from student where id = 3;
-- 删除 id = 4 且 name = '林俊杰的数据'
delete from student where id = 4 and name = '林俊杰'
-- 删除 id < 3 且 name = '周杰伦的数据'
delete from student where id <= 3 and name = '周杰伦';
-- 删除全部数据
delete from student;
-- 将 id = 3 的name改成111
update student set name = '111' where id = 3;

DQL - 查询 ​

执行顺序

sql
from -> where -> group by和聚合函数 -> select -> order by -> limit
from -> where -> group by和聚合函数 -> select -> order by -> limit

基础查询 ​

基础语法

sql
select 字段列表|* from 表名称 where 条件判断
select 字段列表|* from 表名称 where 条件判断

示例

sql
-- 查询id和name两个列
select id,name from student;
-- 查询id、age和name列
select id,name,age from student;
-- 查询age>30所有的数据
select * from student where age>30;
-- 查询id和name两个列
select id,name from student;
-- 查询id、age和name列
select id,name,age from student;
-- 查询age>30所有的数据
select * from student where age>30;

分组聚合 ​

基础语法

sql
select 必须和by后面的列一样|聚合函数 from 表名称 [where 条件] group by 列
聚合函数有：
- sum(列) 求和
- avg(列) 求平均值
- hin(列) 求最小值
- count(列|*) 求总数
GROUP BY中出现了哪个列，哪个列才能出现在SELECT中的非聚合中
select 必须和by后面的列一样|聚合函数 from 表名称 [where 条件] group by 列
聚合函数有：
- sum(列) 求和
- avg(列) 求平均值
- hin(列) 求最小值
- count(列|*) 求总数
GROUP BY中出现了哪个列，哪个列才能出现在SELECT中的非聚合中

示例

sql
-- 通过IsOfficial分组，在求出每组Percentage的总和
select IsOfficial, sum(Percentage) from countrylanguage group by IsOfficial;

-- 通过IsOfficial分组，在求出每组Percentage的平均值
select IsOfficial, avg(Percentage) from countrylanguage group by IsOfficial;

-- 一个sql可以写多个聚合函数
select IsOfficial, avg(Percentage),sum(Percentage),count(*) from countrylanguage group by IsOfficial;
-- 通过IsOfficial分组，在求出每组Percentage的总和
select IsOfficial, sum(Percentage) from countrylanguage group by IsOfficial;

-- 通过IsOfficial分组，在求出每组Percentage的平均值
select IsOfficial, avg(Percentage) from countrylanguage group by IsOfficial;

-- 一个sql可以写多个聚合函数
select IsOfficial, avg(Percentage),sum(Percentage),count(*) from countrylanguage group by IsOfficial;

排序分页 ​

语法

sql
select 列|聚合函数|* from 表名称 where... group by... order by...[asc正序|desc倒序] limit n几条数据[,m  n条数据之后获取m条]
select 列|聚合函数|* from 表名称 where... group by... order by...[asc正序|desc倒序] limit n几条数据[,m  n条数据之后获取m条]

示例

sql
-- Percentage>20然后降序排序
select * from countrylanguage where Percentage >20 order by Percentage desc;

-- 获取20条数据
select * from countrylanguage where Percentage >20 order by Percentage desc limit 20;

-- 从20条数据往后获取5条数据
select * from countrylanguage where Percentage >20 order by Percentage desc limit 20,5;
-- Percentage>20然后降序排序
select * from countrylanguage where Percentage >20 order by Percentage desc;

-- 获取20条数据
select * from countrylanguage where Percentage >20 order by Percentage desc limit 20;

-- 从20条数据往后获取5条数据
select * from countrylanguage where Percentage >20 order by Percentage desc limit 20,5;

开发中运用 ​

安装 ​

pip install pymysql
pip install pymysql

运用 ​

python
from pymysql import Connection
# 连接
conn = Connection(
    host = 'localhost', # 主机
    port = 3306, # 端口
    user = 'root', # 用户名
    password = '123456', # 密码
    autocommit = True, # 自动提交
)
# 获取游标
cursor = conn.cursor()
# 选择数据库
conn.select_db('world')
# 执行sql创建表
cursor.execute('create table country(id int,name varchar(20))')
# 执行sql查询
cursor.execute('select * from country')
# 获取查询结果
res = cursor.fetchall()
for row in res:
    print(row)
# 执行sql插入
cursor.execute("insert into text_py_sql values(2,'周杰伦2',22,'男')")
# 提交  如果 autocommit = True 就不需要commit了
conn.commit()
# 关闭
conn.close()
from pymysql import Connection
# 连接
conn = Connection(
    host = 'localhost', # 主机
    port = 3306, # 端口
    user = 'root', # 用户名
    password = '123456', # 密码
    autocommit = True, # 自动提交
)
# 获取游标
cursor = conn.cursor()
# 选择数据库
conn.select_db('world')
# 执行sql创建表
cursor.execute('create table country(id int,name varchar(20))')
# 执行sql查询
cursor.execute('select * from country')
# 获取查询结果
res = cursor.fetchall()
for row in res:
    print(row)
# 执行sql插入
cursor.execute("insert into text_py_sql values(2,'周杰伦2',22,'男')")
# 提交  如果 autocommit = True 就不需要commit了
conn.commit()
# 关闭
conn.close()

SQLAlchemy 2.0 ​

ORM（对象关系映射）框架，Python 操作数据库的首选方案。

安装 ​

bash
pip install sqlalchemy
pip install sqlalchemy[asyncio]   # 异步支持
pip install psycopg2-binary       # PostgreSQL 驱动
pip install pymysql               # MySQL 驱动
pip install aiosqlite             # SQLite 异步驱动
pip install sqlalchemy
pip install sqlalchemy[asyncio]   # 异步支持
pip install psycopg2-binary       # PostgreSQL 驱动
pip install pymysql               # MySQL 驱动
pip install aiosqlite             # SQLite 异步驱动

核心概念（2.0 风格） ​

python
from sqlalchemy import create_engine, text, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.orm import relationship
from datetime import date

# 1. 定义模型
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False)
    age: Mapped[int | None]  # Optional
    created_at: Mapped[date] = mapped_column(default=date.today)

    # 关系
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    user_id: Mapped[int] = mapped_column(foreign_key="users.id")
    author: Mapped["User"] = relationship(back_populates="posts")

# 2. 创建引擎
engine = create_engine("sqlite:///example.db", echo=True)

# 3. 创建表
Base.metadata.create_all(engine)

# 4. CRUD 操作
with Session(engine) as session:
    # 创建
    user = User(name="Alice", age=30)
    session.add(user)
    session.commit()

    # 查询（2.0 风格）
    stmt = select(User).where(User.name == "Alice")
    result = session.execute(stmt)
    user = result.scalar_one()

    # 更新
    user.age = 31
    session.commit()

    # 删除
    session.delete(user)
    session.commit()
from sqlalchemy import create_engine, text, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, Session
from sqlalchemy.orm import relationship
from datetime import date

# 1. 定义模型
class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(nullable=False)
    age: Mapped[int | None]  # Optional
    created_at: Mapped[date] = mapped_column(default=date.today)

    # 关系
    posts: Mapped[list["Post"]] = relationship(back_populates="author")

class Post(Base):
    __tablename__ = "posts"

    id: Mapped[int] = mapped_column(primary_key=True)
    title: Mapped[str]
    user_id: Mapped[int] = mapped_column(foreign_key="users.id")
    author: Mapped["User"] = relationship(back_populates="posts")

# 2. 创建引擎
engine = create_engine("sqlite:///example.db", echo=True)

# 3. 创建表
Base.metadata.create_all(engine)

# 4. CRUD 操作
with Session(engine) as session:
    # 创建
    user = User(name="Alice", age=30)
    session.add(user)
    session.commit()

    # 查询（2.0 风格）
    stmt = select(User).where(User.name == "Alice")
    result = session.execute(stmt)
    user = result.scalar_one()

    # 更新
    user.age = 31
    session.commit()

    # 删除
    session.delete(user)
    session.commit()

关系查询 ​

python
from sqlalchemy import select
from sqlalchemy.orm import joinedload, selectinload

with Session(engine) as session:
    # 基本查询
    users = session.execute(select(User)).scalars().all()

    # 条件查询
    stmt = select(User).where(
        User.age >= 18,
        User.name.like("%张%"),
    )
    users = session.execute(stmt).scalars().all()

    # 关联查询（N+1 问题解决）
    stmt = select(User).options(selectinload(User.posts))
    users = session.execute(stmt).scalars().all()
    for user in users:
        print(user.posts)  # 不会产生额外查询

    # 聚合查询
    from sqlalchemy import func
    stmt = select(User.age, func.count(User.id)).group_by(User.age)
    results = session.execute(stmt).all()

    # 分页
    stmt = select(User).limit(10).offset(20)
    users = session.execute(stmt).scalars().all()
from sqlalchemy import select
from sqlalchemy.orm import joinedload, selectinload

with Session(engine) as session:
    # 基本查询
    users = session.execute(select(User)).scalars().all()

    # 条件查询
    stmt = select(User).where(
        User.age >= 18,
        User.name.like("%张%"),
    )
    users = session.execute(stmt).scalars().all()

    # 关联查询（N+1 问题解决）
    stmt = select(User).options(selectinload(User.posts))
    users = session.execute(stmt).scalars().all()
    for user in users:
        print(user.posts)  # 不会产生额外查询

    # 聚合查询
    from sqlalchemy import func
    stmt = select(User.age, func.count(User.id)).group_by(User.age)
    results = session.execute(stmt).all()

    # 分页
    stmt = select(User).limit(10).offset(20)
    users = session.execute(stmt).scalars().all()

异步 ORM（asyncio） ​

python
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

# 异步引擎
async_engine = create_async_engine("sqlite+aiosqlite:///example.db")

async def get_users():
    async with AsyncSession(async_engine) as session:
        stmt = select(User)
        result = await session.execute(stmt)
        return result.scalars().all()
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy import select

# 异步引擎
async_engine = create_async_engine("sqlite+aiosqlite:///example.db")

async def get_users():
    async with AsyncSession(async_engine) as session:
        stmt = select(User)
        result = await session.execute(stmt)
        return result.scalars().all()

原生 SQL 执行 ​

python
from sqlalchemy import text

with Session(engine) as session:
    # 查询
    result = session.execute(text("SELECT * FROM users WHERE age > :age"), {"age": 18})
    rows = result.all()

    # 直接执行 DML
    session.execute(
        text("INSERT INTO users (name, age) VALUES (:name, :age)"),
        [{"name": "Bob", "age": 25}, {"name": "Carol", "age": 28}],
    )
    session.commit()
from sqlalchemy import text

with Session(engine) as session:
    # 查询
    result = session.execute(text("SELECT * FROM users WHERE age > :age"), {"age": 18})
    rows = result.all()

    # 直接执行 DML
    session.execute(
        text("INSERT INTO users (name, age) VALUES (:name, :age)"),
        [{"name": "Bob", "age": 25}, {"name": "Carol", "age": 28}],
    )
    session.commit()

数据库迁移（Alembic） ​

bash
pip install alembic

# 初始化
alembic init alembic

# 修改 alembic.ini 中的 sqlalchemy.url
# sqlalchemy.url = sqlite:///example.db

# 生成迁移脚本
alembic revision --autogenerate -m "add users table"

# 执行迁移
alembic upgrade head

# 回滚
alembic downgrade -1

# 查看历史
alembic history
pip install alembic

# 初始化
alembic init alembic

# 修改 alembic.ini 中的 sqlalchemy.url
# sqlalchemy.url = sqlite:///example.db

# 生成迁移脚本
alembic revision --autogenerate -m "add users table"

# 执行迁移
alembic upgrade head

# 回滚
alembic downgrade -1

# 查看历史
alembic history
python
# alembic/env.py 中配置
from models import Base  # 导入你的 Base
target_metadata = Base.metadata
# alembic/env.py 中配置
from models import Base  # 导入你的 Base
target_metadata = Base.metadata

Copyright © 2024-present xiaocao