【Python使用】嘿马头条项目从到完整开发教程第3篇：数据库,SQLAlchemy操作【附代码文档】

举报
程序员一诺python 发表于 2025/07/13 19:49:59 2025/07/13
【摘要】 2. 目标 ToutiaoWeb虚拟机使用说明 Pycharm远程开发 产品与开发 1 产品介绍 2 原型图与UI图 3 技术架构 4 开发 数据库 APScheduler定时任务 定时修正统计数据 RPC简介 1. 什么是RPC 2. 背景与用途 3. 概念说明 4. 优缺点 gRPC 简介 架构 使用方法 Protocol Buffers 1 文档结构 2 注释 3 数据类

教程总体简介：ToutiaoWeb虚拟机使用说明、Pycharm远程开发、产品与开发、数据库、数据库设计、APScheduler定时任务、定时修正统计数据、RPC简介、RPC结构、gRPC、简介、RPC、Protocol Buffers、推荐系统接口定义、补全服务端、编写客户端、头条首页新闻推荐接口编写、即时通讯简介、WebSocket、即时通讯、Socket.IO、头条聊天服务实现、头条在线消息推送实现、Elasticsearch、简介与原理、概念与集群、IK中文分析器、索引与类型、文档、Logstash导入数据、查询、头条全文检索实现、联想提示、头条suggest查询实现、单元测试、Gunicorn、Supervisor、部署相关、理解ORM、SQLAlchemy映射构建、SQLAlchemy操作、数据库理论、分布式ID、数据库优化、Redis、Git工用流、Gitflow工作流、Git总结、头条项目目录、调试方法、JWT & JWS & JWE、JWT的Python库、头条项目实施方案、JWT禁用问题、JWT认证方案、OSS对象存储、CDN、缓存、缓存架构、缓存数据、缓存有效期与淘汰策略、缓存模式、缓存问题、头条项目缓存与存储设计、头条项目缓存实现、项目Redis持久存储实现、APScheduler使用

https://bbs.huaweicloud.com/blogs/453687

https://bbs.huaweicloud.com/blogs/455384

全套教程部分目录：

数据库

  • 数据库设计

  • SQLAlchemy

  • 数据库理论

  • 分布式ID

  • Redis

SQLAlchemy操作

1 新增

1
2
3
4
5
6
user = User(mobile='15612345678', name='itcast')
db.session.add(user)
db.session.commit()
profile = Profile(id=user.id)
db.session.add(profile)
db.session.commit()

对于批量添加也可使用如下语法

1
2
3
4
5
6
db.session.add_all([user1, user2, user3])
db.session.commit()

2 查询

all()

查询所有，返回列表

1
2
3
4
5
6
User.query.all()

first()

查询第一个，返回对象

1
2
3
4
5
6
User.query.first()

get()

根据主键ID获取对象，若主键不存在返回None

1
2
3
4
5
6
User.query.get(2)

另一种查询方式

1
2
3
4
5
6
db.session.query(User).all()
db.session.query(User).first()
db.session.query(User).get(2)

filter_by

进行过虑

1
2
3
4
5
6
User.query.filter_by(mobile='13911111111').first()
User.query.filter_by(mobile='13911111111', id=1).first()  # and关系

filter

进行过虑

1
2
3
4
5
6
User.query.filter(User.mobile=='13911111111').first()

逻辑或

1
2
3
4
5
6
from sqlalchemy import or_
User.query.filter(or_(User.mobile=='13911111111', User.name.endswith('号'))).all()

逻辑与

1
2
3
4
5
6
from sqlalchemy import and_
User.query.filter(and_(User.name != '13911111111', User.mobile.startswith('185'))).all()

逻辑非

1
2
3
4
5
6
from sqlalchemy import not_
User.query.filter(not_(User.mobile == '13911111111')).all()

offset

偏移，起始位置

1
2
3
4
5
6
User.query.offset(2).all()

limit

获取限制数据

1
2
3
4
5
6
User.query.limit(3).all()

order_by

排序

1
2
3
4
5
6
User.query.order_by(User.id).all()  # 正序
User.query.order_by(User.id.desc()).all()  # 倒序

复合查询

1
2
3
4
5
6
User.query.filter(User.name.startswith('13')).order_by(User.id.desc()).offset(2).limit(5).all()

1
2
3
4
5
6
query = User.query.filter(User.name.startswith('13'))
query = query.order_by(User.id.desc())
query = query.offset(2).limit(5)
ret = query.all()

优化查询

1
2
3
4
5
6
user = User.query.filter_by(id=1).first()  # 查询所有字段
select user_id, mobile......

select * from   # 程序不要使用
select user_id, mobile,.... # 查询指定字段

from sqlalchemy.orm import load_only
User.query.options(load_only(User.name, User.mobile)).filter_by(id=1).first() # 查询特定字段

聚合查询

1
2
3
4
5
6
from sqlalchemy import func

db.session.query(Relation.user_id, func.count(Relation.target_user_id)).filter(Relation.relation == Relation.RELATION.FOLLOW).group_by(Relation.user_id).all()

关联查询

1. 使用ForeignKey
1
2
3
4
5
6
class User(db.Model):
    ...
    profile = db.relationship('UserProfile', uselist=False)
    followings = db.relationship('Relation')

class UserProfile(db.Model):
    id = db.Column('user_id', db.Integer, db.ForeignKey('user_basic.user_id'), primary_key=True,  doc='用户ID')
    ...

class Relation(db.Model):
    user_id = db.Column(db.Integer, db.ForeignKey('user_basic.user_id'), doc='用户ID')
    ...



# 测试   


user = User.query.get(1)
user.profile.gender
user.followings
2. 使用primaryjoin
1
2
3
4
5
6
class User(db.Model):
    ...

    profile = db.relationship('UserProfile', primaryjoin='User.id==foreign(UserProfile.id)', uselist=False)
    followings = db.relationship('Relation', primaryjoin='User.id==foreign(Relation.user_id)')



# 测试


user = User.query.get(1)
user.profile.gender
user.followings
3. 指定字段关联查询
1
2
3
4
5
6
class Relation(db.Model):
    ...
    target_user = db.relationship('User', primaryjoin='Relation.target_user_id==foreign(User.id)', uselist=False)

from sqlalchemy.orm import load_only, contains_eager

Relation.query.join(Relation.target_user).options(load_only(Relation.target_user_id), contains_eager(Relation.target_user).load_only(User.name)).all()

3 更新

  • 方式一
1
2
3
4
5
6
user = User.query.get(1)
  user.name = 'Python'
  db.session.add(user)
  db.session.commit()
  • 方式二
1
2
3
4
5
6
User.query.filter_by(id=1).update({'name':'python'})
  db.session.commit()

4 删除

  • 方式一
1
2
3
4
5
6
user = User.query.order_by(User.id.desc()).first()
  db.session.delete(user)
  db.session.commit()
  • 方式二
1
2
3
4
5
6
User.query.filter(User.mobile='18512345678').delete()
  db.session.commit()

5 事务

1
2
3
4
5
6
environ = {'wsgi.version':(1,0), 'wsgi.input': '', 'REQUEST_METHOD': 'GET', 'PATH_INFO': '/', 'SERVER_NAME': 'itcast server', 'wsgi.url_scheme': 'http', 'SERVER_PORT': '80'}

with app.request_context(environ):
    try:
        user = User(mobile='18911111111', name='itheima')
        db.session.add(user)
        db.session.flush() # 将db.session记录的sql传到数据库中执行
        profile = UserProfile(id=user.id)
        db.session.add(profile)
        db.session.commit()
    except:
        db.session.rollback()

数据库

  • 数据库设计

  • SQLAlchemy

  • 数据库理论

  • 分布式ID

  • Redis

数据库理论

1. 复制集与分布式

  • 复制集（Replication）

  • 数据库中数据相同，起到备份作用

  • 高可用 High Available HA

  • 分布式（Distribution）

  • 数据库中数据不同，共同组成完整的数据集合

  • 通常每个节点被称为一个分片（shard)

  • 高吞吐 High Throughput

  • 复制集与分布式可以单独使用，也可以组合使用（即每个分片都组建一个复制集）

  • 关于主（Master）从（Slave）

  • 这个概念是从使用的角度来阐述问题的

  • 主节点 -> 表示程序在这个节点上最先更新数据
  • 从节点 -> 表示这个节点的数据是要通过复制主节点而来
  • 复制集 可选 主从、主主、主主从从
  • 分布式 每个分片都是主，组合使用复制集的时候，复制集的是从

2. MySQL

1） 主从复制

复制分成三步：

  1. master将改变记录到二进制日志(binary log)中（这些记录叫做二进制日志事件，binary log events）；
  2. slave将master的binary log events拷贝到它的中继日志(relay log)；
  3. slave重做中继日志中的事件，将改变反映它自己的数据。

下图描述了这一过程：

该过程的第一部分就是master记录二进制日志。在每个事务更新数据完成之前，master在二日志记录这些改变。MySQL将事务串行的写入二进制日志，即使事务中的语句都是交叉执行的。在事件写入二进制日志完成后，master通知存储引擎提交事务。

下一步就是slave将master的binary log拷贝到它自己的中继日志。首先，slave开始一个工作线程——I/O线程。I/O线程在master上打开一个普通的连接，然后开始binlog dump process。Binlog dump process从master的二进制日志中读取事件，如果已经跟上master，它会睡眠并等待master产生新的事件。I/O线程将这些事件写入中继日志。

SQL slave thread处理该过程的最后一步。SQL线程从中继日志读取事件，更新slave的数据，使其与master中的数据一致。只要该线程与I/O线程保持一致，中继日志通常会位于OS的缓存中，所以中继日志的开销很小。

此外，在master中也有一个工作线程：和其它MySQL的连接一样，slave在master中打开一个连接也会使得master开始一个线程。

利用主从在达到高可用的同时，也可以通过读写分离提供吞吐量。

思考：读写分离对事务是否有影响？

对于写操作包括开启事务和提交或回滚要在一台机器上执行，分散到多台master执行后数据库原生的单机事务就失效了。

对于事务中同时包含读写操作，与事务隔离级别设置有关，如果事务隔离级别为read-uncommitted 或者 read-committed，读写分离没影响，如果隔离级别为repeatable-read、serializable，读写分离就有影响，因为在slave上会看到新数据，而正在事务中的master看不到新数据。

2）分库分表（sharding）

分库分表前的问题

任何问题都是太大或者太小的问题，我们这里面对的数据量太大的问题。

  • 用户请求量太大

因为单服务器TPS，内存，IO都是有限的。 解决方法：分散请求到多个服务器上； 其实用户请求和执行一个sql查询是本质是一样的，都是请求一个资源，只是用户请求还会经过网关，路由，http服务器等。

  • 单库太大

单个数据库处理能力有限；单库所在服务器上磁盘空间不足；单库上操作的IO瓶颈 解决方法：切分成更多更小的库

  • 单表太大

CRUD都成问题；索引膨胀，查询超时 解决方法：切分成多个数据集更小的表。

分库分表的方式方法

一般就是垂直切分和水平切分，这是一种结果集描述的切分方式，是物理空间上的切分。 我们从面临的问题，开始解决，阐述： 首先是用户请求量太大，我们就堆机器搞定（这不是本文重点）。

然后是单个库太大，这时我们要看是因为表多而导致数据多，还是因为单张表里面的数据多。 如果是因为表多而数据多，使用垂直切分，根据业务切分成不同的库。

如果是因为单张表的数据量太大，这时要用水平切分，即把表的数据按某种规则切分成多张表，甚至多个库上的多张表。 分库分表的顺序应该是先垂直分，后水平分。 因为垂直分更简单，更符合我们处理现实世界问题的方式。

垂直拆分
  1. 垂直分表

也就是“大表拆小表”，基于列字段进行的。一般是表中的字段较多，将不常用的， 数据较大，长度较长（比如text类型字段）的拆分到“扩展表“。 一般是针对那种几百列的大表，也避免查询时，数据量太大造成的“跨页”问题。

  1. 垂直分库

垂直分库针对的是一个系统中的不同业务进行拆分，比如用户User一个库，商品Producet一个库，订单Order一个库。 切分后，要放在多个服务器上，而不是一个服务器上。为什么？ 我们想象一下，一个购

【声明】本内容来自华为云开发者社区博主，不代表华为云及华为云开发者社区的观点和立场。转载时必须标注文章的来源（华为云社区）、文章链接、文章作者等基本信息，否则作者和本社区有权追究责任。如果您发现本社区中有涉嫌抄袭的内容，欢迎发送邮件进行举报，并提供相关证据，一经查实，本社区将立刻删除涉嫌侵权内容，举报邮箱： cloudbbs@huaweicloud.com
Python 后端
  • 点赞
  • 收藏
  • 关注作者

评论（0）

0/1000
抱歉，系统识别当前为高风险访问，暂不支持该操作

全部回复

上滑加载中

设置昵称

在此一键设置昵称，即可参与社区互动！

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

*长度不超过10个汉字或20个英文字符，设置后3个月内不可修改。

确认 取消

加入云驻计划，成为创作者

  • 华为云周边好礼
  • 免费体验产品
  • 特殊身份标识
  • 线下官方门票
  • 内部专家零距离
  • 与10000+优质创作者共同成长
立即加入