超越Python：下一步该学什么编程语言？

Ne0inhk

— 5 min read

SQLAlchemy是Python中最流行的ORM（对象关系映射）框架之一，它提供了高效且灵活的数据库操作方式。本文将介绍如何使用SQLAlchemy ORM进行数据库操作。

目录

  1. 安装SQLAlchemy
  2. 核心概念
  3. 连接数据库
  4. 定义数据模型
  5. 创建数据库表
  6. 基本CRUD操作
  7. 查询数据
  8. 关系操作
  9. 事务管理
  10. 最佳实践

安装

bash

pip install sqlalchemy

如果需要连接特定数据库，还需安装相应的驱动程序：

bash

# PostgreSQL pip install psycopg2-binary # MySQL pip install mysql-connector-python # SQLite (Python标准库已包含，无需额外安装)

核心概念

  • Engine：数据库连接的引擎，负责与数据库通信
  • Session：数据库会话，管理所有持久化操作
  • Model：数据模型类，对应数据库中的表
  • Query：查询对象，用于构建和执行数据库查询

连接数据库

python

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker # 创建数据库连接引擎 # SQLite示例 engine = create_engine('sqlite:///example.db', echo=True) # PostgreSQL示例 # engine = create_engine('postgresql://username:password@localhost:5432/mydatabase') # MySQL示例 # engine = create_engine('mysql+mysqlconnector://username:password@localhost:3306/mydatabase') # 创建会话工厂 SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) # 创建会话实例 session = SessionLocal()

定义数据模型

python

from sqlalchemy import Column, Integer, String, ForeignKey from sqlalchemy.orm import relationship, declarative_base # 创建基类 Base = declarative_base() class User(Base): __tablename__ = 'users' id = Column(Integer, primary_key=True, index=True) name = Column(String(50), nullable=False) email = Column(String(100), unique=True, index=True) # 定义一对多关系 posts = relationship("Post", back_populates="author") class Post(Base): __tablename__ = 'posts' id = Column(Integer, primary_key=True, index=True) title = Column(String(100), nullable=False) content = Column(String(500)) author_id = Column(Integer, ForeignKey('users.id')) # 定义多对一关系 author = relationship("User", back_populates="posts") # 定义多对多关系（通过关联表） tags = relationship("Tag", secondary="post_tags", back_populates="posts") class Tag(Base): __tablename__ = 'tags' id = Column(Integer, primary_key=True, index=True) name = Column(String(30), unique=True, nullable=False) posts = relationship("Post", secondary="post_tags", back_populates="tags") # 关联表（用于多对多关系） class PostTag(Base): __tablename__ = 'post_tags' post_id = Column(Integer, ForeignKey('posts.id'), primary_key=True) tag_id = Column(Integer, ForeignKey('tags.id'), primary_key=True)

创建数据库表

python

# 创建所有表 Base.metadata.create_all(bind=engine) # 删除所有表 # Base.metadata.drop_all(bind=engine)

基本CRUD操作

创建数据

python

# 创建新用户 new_user = User(name="张三", email="[email protected]") session.add(new_user) session.commit() # 批量创建 session.add_all([ User(name="李四", email="[email protected]"), User(name="王五", email="[email protected]") ]) session.commit()

读取数据

python

# 获取所有用户 users = session.query(User).all() # 获取第一个用户 first_user = session.query(User).first() # 根据ID获取用户 user = session.query(User).get(1)

更新数据

python

# 查询并更新 user = session.query(User).get(1) user.name = "张三四" session.commit() # 批量更新 session.query(User).filter(User.name.like("张%")).update({"name": "张氏"}, synchronize_session=False) session.commit()

删除数据

python

# 查询并删除 user = session.query(User).get(1) session.delete(user) session.commit() # 批量删除 session.query(User).filter(User.name == "李四").delete(synchronize_session=False) session.commit()

查询数据

基本查询

python

# 获取所有记录 users = session.query(User).all() # 获取特定字段 names = session.query(User.name).all() # 排序 users = session.query(User).order_by(User.name.desc()).all() # 限制结果数量 users = session.query(User).limit(10).all() # 偏移量 users = session.query(User).offset(5).limit(10).all()

过滤查询

python

from sqlalchemy import or_ # 等值过滤 user = session.query(User).filter(User.name == "张三").first() # 模糊查询 users = session.query(User).filter(User.name.like("张%")).all() # IN查询 users = session.query(User).filter(User.name.in_(["张三", "李四"])).all() # 多条件查询 users = session.query(User).filter( User.name == "张三", User.email.like("%@example.com") ).all() # 或条件 users = session.query(User).filter( or_(User.name == "张三", User.name == "李四") ).all() # 不等于 users = session.query(User).filter(User.name != "张三").all()

聚合查询

python

from sqlalchemy import func # 计数 count = session.query(User).count() # 分组计数 user_post_count = session.query( User.name, func.count(Post.id) ).join(Post).group_by(User.name).all() # 求和、平均值等 avg_id = session.query(func.avg(User.id)).scalar()

连接查询

python

# 内连接 results = session.query(User, Post).join(Post).filter(Post.title.like("%Python%")).all() # 左外连接 results = session.query(User, Post).outerjoin(Post).all() # 指定连接条件 results = session.query(User, Post).join(Post, User.id == Post.author_id).all()

关系操作

python

# 创建带关系的对象 user = User(name="赵六", email="[email protected]") post = Post(title="我的第一篇博客", content="Hello World!", author=user) session.add(post) session.commit() # 通过关系访问 print(f"文章 '{post.title}' 的作者是 {post.author.name}") print(f"用户 {user.name} 的所有文章:") for p in user.posts: print(f" - {p.title}") # 多对多关系操作 python_tag = Tag(name="Python") sqlalchemy_tag = Tag(name="SQLAlchemy") post.tags.append(python_tag) post.tags.append(sqlalchemy_tag) session.commit() print(f"文章 '{post.title}' 的标签:") for tag in post.tags: print(f" - {tag.name}")

事务管理

python

# 自动提交事务 try: user = User(name="测试用户", email="[email protected]") session.add(user) session.commit() except Exception as e: session.rollback() print(f"发生错误: {e}") # 使用事务上下文管理器 from sqlalchemy.orm import Session def create_user(session: Session, name: str, email: str): try: user = User(name=name, email=email) session.add(user) session.commit() return user except: session.rollback() raise # 嵌套事务 with session.begin_nested(): user = User(name="事务用户", email="[email protected]") session.add(user) # 保存点 savepoint = session.begin_nested() try: user = User(name="保存点用户", email="[email protected]") session.add(user) savepoint.commit() except: savepoint.rollback()

最佳实践

  1. 会话管理：为每个请求创建新会话，请求结束后关闭
  2. 异常处理：始终处理异常并适当回滚事务
  3. 延迟加载：注意N+1查询问题，使用 eager loading 优化
  4. 连接池：合理配置连接池大小和超时设置
  5. 数据验证：在模型层或应用层验证数据完整性

python

# 使用上下文管理器管理会话 from contextlib import contextmanager @contextmanager def get_db(): db = SessionLocal() try: yield db db.commit() except Exception: db.rollback() raise finally: db.close() # 使用示例 with get_db() as db: user = User(name="上下文用户", email="[email protected]") db.add(user)

总结

SQLAlchemy ORM提供了强大而灵活的数据库操作方式，通过本文的介绍，您应该能够：

  1. 安装和配置SQLAlchemy
  2. 定义数据模型和关系
  3. 执行基本的CRUD操作
  4. 构建复杂查询
  5. 管理数据库事务
  6. 遵循最佳实践

SQLAlchemy还有更多高级特性，如混合属性、事件监听、自定义查询等，值得进一步探索学习。

Read more

华为OD机试真题2025双机位C卷 Python&JS 实现【自动泊车】

华为OD机试真题2025双机位C卷 Python&JS 实现【自动泊车】

目录 题目 思路 Code 题目 题目描述 在某商场的地下停车场，部署了一套智能导航系统。停车场可以看作是一个 r*c 的网格矩阵，其中：0 表示该位置是空的行车道，车辆可以通行。1 表示该位置存有障碍物、立柱或其他已停放的车辆，车辆无法通行。 停车场的入口统一设在坐标 [0, 0] 处。现在有一辆车进入停车场，需要前往指定的目标车位 [m, n]。 车辆在停车场内只能沿着上、下、左、右四个方向移动，每移动一个格子计为步数 1。请你帮车主规划一条从入口到目标车位的最短路径。输入描述第一行输入两个整数 m 和 n，表示目标车位的行下标和列下标。第二行输入两个整数 row 和 col，表示停车场的总行数和总列数。接下来的 row 行，每行包含 col

By Ne0inhk
Python 3.12 logging - 07 - LogRecord

Python 3.12 logging - 07 - LogRecord

LogRecord的基本概念 LogRecord 是 Python logging 模块中代表一条日志事件的数据容器。简单来说，它就像一张“记录单”，每当程序调用日志方法（如 logger.info()）时，logging 会自动创建一个 LogRecord 对象，并把所有相关信息都装进去，包括：日志消息内容、日志级别（DEBUG/INFO 等）、产生日志的时间戳、源代码位置（文件名、行号、函数名）、当前线程/进程 ID、异常信息（如果有）、其他自定义属性（通过 extra 添加）。 这个“记录单”随后会被传递给日志处理器（Handler）和格式化器（Formatter），最终被转换成我们看到的输出文本。 一、LogRecord类机制解析 LogRecord是logging模块的核心类，用于封装日志事件的所有信息。

By Ne0inhk

涛哥聊Python | 程序员必看：Codex 和 Claude Code 实战对比，差别比你想的更大！

本文来源公众号“涛哥聊Python”，仅用于学术分享，侵权删，干货满满。 原文链接：https://mp.weixin.qq.com/s/NPzwT-5_qt9ncWxYaaQpYg 程序开发，往往不只是思考逻辑，更多时间消耗在那些重复又琐碎的环节，接口需要写一堆模板代码，参数的小改动要牵连多个文件，修个 bug 还得来回补测试，这些工作不难，但却很耗时。 正因为如此，AI 编程助手逐渐进入开发者的日常，它们虽然不能完全替代人类思考，却能帮我们把重复的部分自动化。 在众多工具中，Codex 和 Claude Code 是讨论度最高的两个，一个专注于把自然语言快速翻译成代码，另一个则成为项目里的智能合作者，这两个工具的功能定位不相同，开发者可以根据自己的需求来选择最合适的助手。 Codex：从“人话”到“代码”的翻译官 Codex 的设计思路很直接：把自然语言转化为代码，只要用一句需求，它就能生成相应的实现，

By Ne0inhk
Ubuntu系统下Python连接国产KingbaseES数据库实现增删改查

Ubuntu系统下Python连接国产KingbaseES数据库实现增删改查

摘要：本文将介绍Ubuntu系统下如何使用Python连接国产金仓数据库KingbaseES，并实现基本的增删改查操作。文中将通过具体代码示例展示连接数据库、执行SQL语句以及处理结果的全过程。这里把Python连接KingbaseES的经验整理一下，希望能帮到同样踩坑的兄弟。 目录 1.环境准备与驱动安装 1.1 科普ksycopg2知识 1.2 官方下载ksycopg2驱动 1.3 安装ksycopg2驱动 2. 连接KingbaseES数据库 3. 创建数据表 4. 实现增删改查功能 4.1 新增 4.2 查询 4.3 修改 4.4 删除 4.5 封装一个类crud方便复用 5.总结 1.环境准备与驱动安装 KingbaseES提供了专门的Python驱动包ksycopg2，它是基于Python DB API 2.0规范实现的线程安全数据库适配器！

By Ne0inhk