Python 使用SQLAlchemy数据库模块

SQLAlchemy 是用Python编程语言开发的一个开源项目，它提供了SQL工具包和ORM对象关系映射工具，使用MIT许可证发行，SQLAlchemy 提供高效和高性能的数据库访问，实现了完整的企业级持久模型。

ORM（对象关系映射）是一种编程模式，用于将对象与关系型数据库中的表和记录进行映射，从而实现通过面向对象的方式进行数据库操作。ORM 的目标是在编程语言中使用类似于面向对象编程的语法，而不是使用传统的 SQL 查询语言，来操作数据库。

主要思想是将数据库表的结构映射到程序中的对象，通过对对象的操作来实现对数据库的操作，而不是直接编写 SQL 查询。ORM 工具负责将数据库记录转换为程序中的对象，反之亦然。

ORM 的核心概念包括：

  1. 实体（Entity）： 在 ORM 中，实体是指映射到数据库表的对象。每个实体对应数据库中的一条记录。
  2. 属性（Attribute）： 实体中的属性对应数据库表中的列。每个属性表示一个字段。
  3. 关系（Relationship）： ORM 允许定义实体之间的关系，例如一对多、多对一、多对多等。这种关系会映射到数据库表之间的关系。
  4. 映射（Mapping）： ORM 负责将实体的属性和方法映射到数据库表的列和操作。
  5. 会话（Session）： ORM 提供了会话来管理对象的生命周期，包括对象的创建、更新和删除。
  6. 查询语言： ORM 通常提供一种查询语言，允许开发者使用面向对象的方式编写查询，而不是直接使用 SQL。

对象映射ROM模型可连接任何关系数据库，连接方法大同小异，以下总结了如何连接常用的几种数据库方式。

# sqlite 创建数据库连接
engine = create_engine('sqlite:///database.db', echo=False)# sqlite 创建内存数据库
engine = create_engine('sqlite://')
engine = create_engine('sqlite:///:memory:', echo=True)# PostgreSQL 创建数据库连接
engine = create_engine('postgresql://scott:tiger@localhost/mydatabase')                # default
engine = create_engine('postgresql+psycopg2://scott:tiger@localhost/mydatabase')       # psycopg2
engine = create_engine('postgresql+pg8000://scott:tiger@localhost/mydatabase')         # pg8000# MySQL 创建数据库连接
engine = create_engine('mysql://scott:tiger@localhost/foo')                  # default
engine = create_engine('mysql+mysqldb://scott:tiger@localhost/foo')          # mysql-python
engine = create_engine('mysql+mysqlconnector://scott:tiger@localhost/foo')   # MySQL-connector-python
engine = create_engine('mysql+oursql://scott:tiger@localhost/foo')           # OurSQL# Oracle 创建数据库连接
engine = create_engine('oracle://scott:tiger@127.0.0.1:1521/sidname')
engine = create_engine('oracle+cx_oracle://scott:tiger@tnsname')# MSSQL 创建数据库连接
engine = create_engine('mssql+pyodbc://scott:tiger@mydsn')                   # pyodbc
engine = create_engine('mssql+pymssql://scott:tiger@hostname:port/dbname')   # pymssql

数据表创建

简单的创建一个User映射类，映射到UserDB库上，分别增加几个常用的数据库字段，并插入一些测试数据。

import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text# 建立基本映射类
Base = declarative_base()# 创建SQLITE数据库
engine = create_engine('sqlite:///:memory:', echo=False)# 创建映射类User
class User(Base):__tablename__ = 'UserDB'# 主键 primary_key | 自动增长 autoincrement | 不为空 nullable | 唯一性约束 uniqueid = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)# 字符串类型username = Column(String(32), nullable=True, default="none")password = Column(String(32), nullable=True, default="none")# 姓名字段默认值是0age = Column(Integer,nullable=False, default=0)# 增加创建日期 [日期:时间]create_time = Column(DateTime, default=datetime.datetime.now)# onupdate=datetime.now 每次更新数据的时候都要更新该字段值update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)# 增加用户分数user_value = Column(Float, default=0.0)# 枚举类型定义# tag = Column(Enum("python",'flask','django'))# __repr__方法用于输出该类的对象被print()时输出的字符串，如果不想写可以不写def __repr__(self):return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)if __name__ == "__main__":print("当前表名: {}".format(User.__table__))# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表Base.metadata.create_all(engine, checkfirst=True)# 逐条增加新记录insert_user = User(username='lyshark', password='123456', age=24, user_value=12.5)session.add(insert_user)insert_user = User(username='sqlalchemy', password='123', age=34, user_value=45.8)session.add(insert_user)# 插入多条记录session.add_all([User(username="admin", password="123123", age=54, user_value=66.9),User(username="root", password="3456576", age=67, user_value=98.4)])# 提交事务session.commit()

数据库查询

演示了通过ORM关系映射实现对单表的简单查询与筛选过滤功能。

import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime# 建立基本映射类
Base = declarative_base()# 创建SQLITE数据库
engine = create_engine('sqlite:///:memory:', echo=False)# 创建映射类User
class User(Base):__tablename__ = 'UserDB'id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)username = Column(String(32), nullable=True, default="none")password = Column(String(32), nullable=True, default="none")age = Column(Integer,nullable=False, default=0)create_time = Column(DateTime, default=datetime.datetime.now)update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)user_value = Column(Float, default=0.0)# __repr__方法用于输出该类的对象被print()时输出的字符串，如果不想写可以不写def __repr__(self):return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)if __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# 查询所有字段all_value = session.query(User).all()for item in all_value:print("ID: {} --> 用户: {}".format(item.id, item.username))# 查询指定字段key_value = session.query(User.username,User.password).all()print(key_value)# 查询第一条first_value = session.query(User).first()print("第一条记录: {} {}".format(first_value.username, first_value.password))# 使用过滤器 [ 过滤出age>20的用户,输出其(id,username)字段 ]filter_value = session.query(User.id,User.username).filter(User.age > 20).all()print("过滤结果: {}".format(filter_value))# 排序输出 [ 正序/倒序 ]sort_value = session.query(User.username,User.age).order_by(User.age).all()print("正序排列: {}".format(sort_value))sort_value = session.query(User.username,User.age).order_by(User.age.desc()).all()print("倒序排列: {}".format(sort_value))# 查询计数count_value = session.query(User).count()print("记录条数: {}".format(count_value))# and/or 条件过滤 默认为and 在filter()中用,分隔多个条件表示,如果是or则需增加or_连接多个条件and_value = session.query(User.username,User.age).filter(User.age >= 20, User.age <= 40).all()print("与查询: {}".format(and_value))or_value = session.query(User.username,User.age).filter(or_(User.age >= 20, User.age <= 40)).all()print("或查询: {}".format(or_value))# 等于查询equal_value = session.query(User.username,User.password).filter(User.age == 67).all()print("等于查询: {}".format(equal_value))not_equal_value = session.query(User.username,User.password).filter(User.age != 67).all()print("不等于查询: {}".format(not_equal_value))# like模糊匹配like_value = session.query(User.username,User.create_time).filter(User.username.like("%ly%")).all()print("模糊匹配: {}".format(like_value))# in查询范围in_value = session.query(User.username,User.password).filter(User.age.in_([24,34])).all()print("查询两者: {}".format(in_value))not_in_value = session.query(User.username,User.password).filter(User.age.notin_([24,34])).all()print("查询非两者: {}".format(not_in_value))# op正则匹配查询op_value = session.query(User.username).filter(User.username.op("regexp")("^a")).all()print("正则匹配: {}".format(op_value))# 调用数据库内置函数func_value = session.query(func.count(User.age)).one()print("调用函数: {}".format(func_value))# 数据切片cat_value = session.query(User.username).all()[:2]print("输出前两条: {}".format(cat_value))cat_value = session.query(User.username).offset(5).limit(3).all()print("第6行开始显示前3个: {}".format(cat_value))cat_value = session.query(User.username).order_by(User.id.desc())[0:10]print("输出最后10个: {}".format(cat_value))# 非空查询isnot_value = session.query(User).filter(User.username.isnot(None)).all()print("非空显示: {}".format(isnot_value))null_value = session.query(User).filter(User.username.is_(None)).all()print("为空显示: {}".format(null_value))# 分组测试group_by = session.query(User.username,func.count(User.id)).group_by(User.age).all()print("根据年龄分组: {}".format(group_by))# 进一步过滤查询having_by = session.query(User.username,User.age).group_by(User.age).having(User.age > 30).all()print("以age分组,并查询age大于30的记录: {}".format(having_by))

数据库修改

演示了修改数据库参数以及对数据库指定记录的删除功能。

import sqlite3,time,datetime
from sqlalchemy import func
from sqlalchemy import or_
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime# 建立基本映射类
Base = declarative_base()# 创建SQLITE数据库
engine = create_engine('sqlite:///:memory:', echo=False)# 创建映射类User
class User(Base):__tablename__ = 'UserDB'id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)username = Column(String(32), nullable=True, default="none")password = Column(String(32), nullable=True, default="none")age = Column(Integer,nullable=False, default=0)create_time = Column(DateTime, default=datetime.datetime.now)update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)user_value = Column(Float, default=0.0)# __repr__方法用于输出该类的对象被print()时输出的字符串，如果不想写可以不写def __repr__(self):return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)if __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# 修改数据: 先查询在修改select_update = session.query(User).filter_by(username="lyshark").first()select_update.password = "test1234"session.commit()# 修改数据: 直接修改session.query(User).filter_by(username="lyshark").update({User.password: 'abcd'})session.commit()session.query(User).filter_by(username="lyshark").update({"password": '123456'})session.commit()# 删除数据: 先查询在删除del_ptr = session.query(User).filter_by(username="lyshark").first()session.delete(del_ptr)session.commit()# 删除数据: 直接删除session.query(User).filter(User.username=="sqlalchemy").delete()session.commit()

数据库查询转字典

将从数据库中过滤查询指定的记录，并将该记录转换为字典或JSON格式，利于解析。

import sqlite3,time,datetime,json
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float,DateTime# 建立基本映射类
Base = declarative_base()# 创建SQLITE数据库
engine = create_engine('sqlite:///:memory:', echo=False)# 创建映射类User
class User(Base):__tablename__ = 'UserDB'id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)username = Column(String(32), nullable=True, default="none")password = Column(String(32), nullable=True, default="none")age = Column(Integer,nullable=False, default=0)create_time = Column(DateTime, default=datetime.datetime.now)update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)user_value = Column(Float, default=0.0)# 查询结果转字典 (保留数据类型)def single_to_dict(self):return {c.name: getattr(self, c.name) for c in self.__table__.columns}# 查询结果转字典 (全转为字符串)def dobule_to_dict(self):result = {}for key in self.__mapper__.c.keys():if getattr(self, key) is not None:result[key] = str(getattr(self, key))else:result[key] = getattr(self, key)return result# 将查询结果转为JSON
def to_json(all_vendors):v = [ ven.dobule_to_dict() for ven in all_vendors ]return vif __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# 查询结果转为字典(保持数据库格式)key_value = session.query(User).first()data = key_value.single_to_dict()print("转为字典: {}".format(data))# 查询结果转为字典(字符串格式)key_value = session.query(User).first()data = key_value.dobule_to_dict()print("转为字符串字典: {}".format(data))# 查询结果转为JSON格式key_value = session.query(User)data = to_json(key_value)print("转为JSON格式: {}".format(data))

数据库类内函数调用

用户在使用ORM模型定义类时，可以同时在该映射类中定义各种针对类模型的处理函数，实现对数据的动态处理

from werkzeug.security import generate_password_hash,check_password_hash
import sqlite3,datetime,time
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text# 创建SQLITE数据库
engine = create_engine("sqlite:///:memory:", encoding='utf-8')
Base = declarative_base()                                       # 生成orm基类# 创建会话
Session_class = sessionmaker(bind=engine)                       # 创建与数据库的会话session
session = Session_class()                                       # 生成session实例# 创建映射类User
class User(Base):__tablename__ = 'UserDB'id = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)username = Column(String(64))_password_hash_ = Column(String(256))                       # 加下划线作为私有函数,无法被外部访问email = Column(String(64))# 设置一个password字段用来设置密码@propertydef password(self):raise Exception("密码不能被读取")# 赋值password字段时,则自动加密存储@password.setterdef password(self, value):self._password_hash_ = generate_password_hash(value)# 使用 check_password,进行密码校验 返回True False。def check_password(self, pasword):return check_password_hash(self._password_hash_, pasword)# 设置输出函数def print_function(self):# 密码不可读调用 self.password 会报错# print("用户: {} 密码: {}".format(self.username,self.password))print("用户: {} email: {}".format(self.username, self.email))return Trueif __name__ == "__main__":print("当前表名: {}".format(User.__table__))# 创建数据表Base.metadata.create_all(engine, checkfirst=True)# 插入测试数据insert = User(username="lyshark",password="123123",email="lyshark@163.com")session.add(insert)insert = User(username="admin",password="556677",email="lyshark@163.com")session.add(insert)session.commit()# 查询测试tag = session.query(User).filter_by(username="lyshark").first()print("测试密码是否正确: {}".format(tag.check_password("123123")))# 调用函数验证当前用户tag = session.query(User).filter_by(username="admin").first()func = tag.print_function()print("输出测试: {}".format(func))

数据库聚合函数

通过func库调用数据库内的聚合函数,实现统计最大最小平均数等数据。

import sqlite3,datetime,time
from sqlalchemy import func
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Float, DateTime, Date, Time, Boolean, DECIMAL, Enum, Text# 建立基本映射类
Base = declarative_base()# 创建SQLITE数据库
engine = create_engine('sqlite:///:memory:', echo=False)# 创建映射类User
class User(Base):__tablename__ = 'UserDB'# 主键 primary_key | 自动增长 autoincrement | 不为空 nullable | 唯一性约束 uniqueid = Column(Integer, primary_key=True, autoincrement=True, nullable=True, unique=True)# 字符串类型username = Column(String(32), nullable=True, default="none")password = Column(String(32), nullable=True, default="none")# 姓名字段默认值是0age = Column(Integer,nullable=False, default=0)# 增加创建日期 [日期:时间]create_time = Column(DateTime, default=datetime.datetime.now)# onupdate=datetime.now 每次更新数据的时候都要更新该字段值update_time = Column(DateTime, onupdate=datetime.datetime.now, default=datetime.datetime.now)# 增加用户分数user_value = Column(Float, default=0.0)# 枚举类型定义# tag = Column(Enum("python",'flask','django'))# __repr__方法用于输出该类的对象被print()时输出的字符串，如果不想写可以不写def __repr__(self):return "<UserDB(username='%s',password='%s')>" % (self.username,self.password)if __name__ == "__main__":print("当前表名: {}".format(User.__table__))# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表Base.metadata.create_all(engine, checkfirst=True)# 统计总数count = session.query(func.count(User.id)).first()print("总记录: {}".format(count))# age 字段平均值age_avg = session.query(func.avg(User.age)).first()print("平均值: {}".format(age_avg))# age 字段最大值age_max = session.query(func.max(User.age)).first()print("最大值: {}".format(age_max))# age 最小值age_min = session.query(func.min(User.age)).one()print("最小值: {}".format(age_min))# age 求和 只求前三个的和age_sum = session.query(func.sum(User.age)).one()print("求总数: {}".format(age_sum))# 提交事务session.commit()

ORM定义一对多关系

SQLAlchemy提供了一个relationship，这个类可以定义属性，以后在访问相关联的表的时候就直接可以通过属性访问的方式就可以访问得到。

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey# 打开数据库
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)# 主表
class Author(Base):__tablename__ = 'author'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(10), nullable=False)# 定义外键关联到Book模型上面,主表是authorbooks = relationship('Book', backref='author')def __repr__(self):return '<Author:(id={}, name={})>'.format(self.id, self.name)# 从表
class Book(Base):__tablename__ = 'book'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(20), nullable=False)# 外键关联到主表author的id字段上author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))def __repr__(self):return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)if __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# ------------------------------------------------------# 创建数据并插入author1 = Author(id=1, name="张三")author2 = Author(id=2, name="李四")book1 = Book(name="<Python 开发>", author_id=1)book2 = Book(name="<C++ 开发教程>", author_id=1)book3 = Book(name="<C# 从入门到精通>", author_id=1)book4 = Book(name="<渗透测试指南>", author_id=2)book5 = Book(name="<nmap 扫描工具指南>", author_id=2)session.add_all([author1,book1,book2,book3])session.add_all([author2,book4,book5])session.commit()# ------------------------------------------------------# 关联插入模式author1 = Author(id=3, name="王五")book1 = Book(name="<Python 开发>", author_id=1)book2 = Book(name="<C++ 开发教程>", author_id=1)book3 = Book(name="<C# 从入门到精通>", author_id=1)author1.books.append(book1)author1.books.append(book2)author1.books.append(book3)session.add(author1)session.commit()# ------------------------------------------------------# 一对多查询测试book = session.query(Book).get(1)print("书籍作者: {}".format(book.author.name))author = session.query(Author).get(1)print("书籍数量: {}".format(len(author.books)))for book_name in author.books:print("书籍: {}".format(book_name.name))

ORM定义一对一关系

如果想要将两个模型映射成一对一的关系，那么应该在父模型中，指定引用的时候，要传递一个uselist=False参数进去。就是告诉父模型，以后引用这个从模型的时候，不再是一个列表了，而是一个对象了。

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey# 打开数据库
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)# 主表
class User(Base):__tablename__ = 'user'id = Column(Integer, primary_key=True, autoincrement=True)username = Column(String(50), nullable=False)def __repr__(self):return 'User(username:%s)' % self.username# 从表
class UserExtend(Base):__tablename__ = 'user_extend'id = Column(Integer,primary_key=True,autoincrement=True)school = Column(String(50))age = Column(String(32))sex = Column(String(32))uid = Column(Integer,ForeignKey('user.id'))user = relationship('User',backref=backref('extend', uselist=False))#uselist=False 告诉父模型 以后引用时不再是列表 而是对象def __repr__(self):return 'extend(school:%s)'%self.schoolif __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# ------------------------------------------------------# 插入测试数据user = User(username="lyshark")extend = UserExtend(school="<家里蹲大学>", age="22", sex="M")extend.user = usersession.add(extend)session.commit()# ------------------------------------------------------# 一对一关系测试user_ptr = session.query(User).first()print("用户名: {} --> 学校: {}".format(user_ptr.username,user_ptr.extend.school))extend_ptr = session.query(UserExtend).first()print("用户名: {} --> 学校: {} --> 年龄: {} --> 性别: {}".format(extend_ptr.user.username,extend_ptr.school,extend_ptr.age,extend_ptr.sex))

ORM定义多对多关系

多对多与上面的一对多，一对一不同，创建多对对必须使用中间表Table来解决查询问题。

  • 多对多的关系需要通过一张中间表来绑定他们之间的关系。
  • 先把两个需要做多对多的模型定义出来
  • 使用Table定义一个中间表，中间表一般就是包含两个模型的外键字段就可以了，并且让他们两个来作为一个“复合主键”。
  • 在两个需要做多对多的模型中随便选择一个模型，定义一个relationship属性，来绑定三者之间的关系，在使用relationship的时候，需要传入一个secondary=中间表。
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker,relationship,backref
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey, Table
from sqlalchemy import Column,INT,VARCHAR,ForeignKey
from sqlalchemy.orm import relationship# 打开数据库
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)# 女孩表
class Girls(Base):__tablename__ = "girl"id = Column(Integer,primary_key=True, autoincrement=True)name = Column(String(32))# 建立多对多关系g2b = relationship("Boys",backref="b2g",secondary="hotel")# 男孩表
class Boys(Base):__tablename__ = "boy"id = Column(Integer,primary_key=True, autoincrement=True)name = Column(String(32))# 映射关系表
class Table(Base):__tablename__ = "hotel"id = Column(Integer, primary_key=True, autoincrement=True)boy_id = Column(Integer,ForeignKey("boy.id"))girl_id = Column(Integer,ForeignKey("girl.id"))if __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# ------------------------------------------------------# 增加数据 - relationship 正向girl_obj = Girls(name="女孩主")girl_obj.g2b = [Boys(name="男孩从1"),Boys(name="男孩从2")]session.add(girl_obj)session.commit()# 增加数据 - relationship 反向boy_obj = Boys(name="男孩主")boy_obj.b2g = [Girls(name="女孩从1"),Girls(name="女孩从2")]session.add(boy_obj)session.commit()# ------------------------------------------------------# 正向查询girl_obj_list = session.query(Girls).all()for girl_obj in girl_obj_list:for boy in girl_obj.g2b:print(girl_obj.name,boy.name)# 反向查询boy_obj_list = session.query(Boys).all()for boy in boy_obj_list:for girl in boy.b2g:print(girl.name,boy.name)

连接查询与子查询

连接查询通过JOIN语句实现，子查询则通过subquery实现，首先需要创建一对多关系然后才可使用子查询。

from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine, Column, Integer, String, Text, func, ForeignKey# 打开数据库
Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)# 主表
class Author(Base):__tablename__ = 'author'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(10), nullable=False)# 定义外键关联到Book模型上面,主表是authorbooks = relationship('Book', backref='author')def __repr__(self):return '<Author:(id={}, name={})>'.format(self.id, self.name)# 从表
class Book(Base):__tablename__ = 'book'id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(20), nullable=False)# 外键关联到主表author的id字段上author_id = Column(Integer, ForeignKey('author.id',ondelete="RESTRICT"))def __repr__(self):return '<Book:(id={}, name={}, author_id={})>'.format(self.id, self.name, self.author_id)if __name__ == "__main__":# 创建会话Session = sessionmaker(bind=engine)session = Session()# 创建数据表(存在则跳过)Base.metadata.create_all(engine, checkfirst=True)# ------------------------------------------------------# 创建数据并插入author1 = Author(id=1, name="张三")author2 = Author(id=2, name="李四")book1 = Book(name="<Python 开发>", author_id=1)book2 = Book(name="<C++ 开发教程>", author_id=1)book3 = Book(name="<C# 从入门到精通>", author_id=1)book4 = Book(name="<渗透测试指南>", author_id=2)book5 = Book(name="<nmap 扫描工具指南>", author_id=2)session.add_all([author1,book1,book2,book3])session.add_all([author2,book4,book5])session.commit()# ------------------------------------------------------# 连表查询no_join_select = session.query(Author).filter(Author.id == Book.id).filter(Author.name == "王五").all()print("查询主键==从键 并且 Author.name == 王五的记录: {}".format(no_join_select))# JOIN 连接查询join = session.query(Author).join(Book).filter(Book.name=="<nmap 扫描工具指南>").first().nameprint("查询主表Author中的Book书名的作者是: {}".format(join))join = session.query(Book).join(Author).filter(Author.name=="李四").all()for book in join:print("查询从表Book中的Author作者有哪些书: {}".format(book.name))# subquery 子查询sbq = session.query(Book.author_id,func.count('*').label("book_count")).group_by(Book.author_id).subquery()print("查询出书籍编号计数(子语句): {}".format(sbq))sub_join = session.query(Author.name,sbq.c.book_count).outerjoin(sbq,Author.id == sbq.c.author_id).all()print("查询用户有几本书(主语句): {}".format(sub_join))

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.mzph.cn/news/163060.shtml

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

MySQL For Windows的下载与安装

MySQL For Windows的下载与安装

教程https://www.bilibili.com/read/cv26499785/ windowse下载地址https://dev.mysql.com/get/Downloads/MySQLInstaller/mysql-installer-community-8.0.35.0.msi
阅读更多...
代理模式 (Proxy Pattern)

代理模式 (Proxy Pattern)

定义&#xff1a; 代理模式&#xff08;Proxy Pattern&#xff09;是一种结构型设计模式&#xff0c;它通过提供一个代理&#xff08;或称代表&#xff09;对象来控制对另一个对象的访问。这种模式创建了一个代理对象&#xff0c;用来代表实际对象的功能&#xff0c;从而可以在…
阅读更多...
spring boot 热部署

spring boot 热部署

相信小伙伴们在日常的开发中&#xff0c;调试代码时&#xff0c;免不了经常修改代码&#xff0c;这个时候&#xff0c;为了验证效果&#xff0c;必须要重启 Spring Boot 应用。 频繁地重启应用&#xff0c;导致开发效率降低&#xff0c;加班随之而来。有没有什么办法&#xff0…
阅读更多...
宏电股份受邀参加中国联通战新共创启航大会，共筑产业生态，链通数智未来

宏电股份受邀参加中国联通战新共创启航大会，共筑产业生态，链通数智未来

11月21日&#xff0c;由中国联通举办的主题为“共筑产业生态&#xff0c;链通数智未来”的网络安全现代产业链共链行动计划暨战新共创启航大会“5G工业互联网”专题供需对接会在北京顺利召开&#xff0c;宏电股份董事长左绍舟应邀出席活动。 会议现场&#xff0c;中国联通雁飞…
阅读更多...
Rust开发——数据对象的内存布局

Rust开发——数据对象的内存布局

枚举与Sized 数据 一般数据类型的布局是其大小&#xff08;size&#xff09;、对齐方式&#xff08;align&#xff09;及其字段的相对偏移量。 1. 枚举&#xff08;Enum&#xff09;的布局&#xff1a; 枚举类型在内存中的布局通常是由编译器来确定的。不同的编译器可能有不…
阅读更多...
centos7 系统keepalived 定时执行脚本

centos7 系统keepalived 定时执行脚本

安装keepalived yum install -y keepalived 修改配置文件 配置文件路径 /etc/keepalived 配置文件内容 global_defs {router_id localhost.localdomain # 访问到主机&#xff0c;本机的hostname&#xff0c;需要修改 }vrrp_script chk_http_port {script "/etc/kee…
阅读更多...
INFLOW：用于检测隐藏服务器的反向网络流水印

INFLOW：用于检测隐藏服务器的反向网络流水印

文章信息 论文题目&#xff1a;INFLOW: Inverse Network Flow Watermarking for Detecting Hidden Servers 期刊&#xff08;会议&#xff09;&#xff1a;IEEE INFOCOM 2018 - IEEE Conference on Computer Communications 级别&#xff1a;CCF A 文章链接&#xff1a;https:…
阅读更多...
Docker 安装 Apache

Docker 安装 Apache

目录 拉取官方 Apache 镜像 查看本地镜像 列出正在运行的容器 运行 Apache 容器 创建一个 HTML 文件&#xff1a;index.html 访问 Apache 拉取官方 Apache 镜像 查找 Docker Hub 上的 httpd 镜像。 可以通过 Tags 查看其他版本的 httpd&#xff0c;默认是最新版本 httpd…
阅读更多...
人工智能学习阶段有哪些?

人工智能学习阶段有哪些?

人工智能学习阶段有哪些? 人工智能是一个跨学科、跨领域的杂交学科&#xff0c;未来的趋势来看,人工智能的出现使人们的生活变得更美好、更便捷&#xff0c;许多小伙伴想学习人工智能&#xff0c;其实看似人工智能比较杂多&#xff0c;无从下手&#xff0c;我们只要从以下7个阶…
阅读更多...
go build自建包报错package XXX is not in std

go build自建包报错package XXX is not in std

在构建包名的时候遇到了如下错误 C:\Users\xxx\Desktop\code\golx\src>go build example package example is not in std (D:\Go\src\example) 解决方法 修改 GO111MODULE go env -w GO111MODULEoff
阅读更多...
SpringMVC 基础知识

SpringMVC 基础知识

学习目标 掌握基于 SpringMVC 获取请求参数与响应 json 数据操作熟练应用基于 REST 风格的请求路径设置与参数传递能够根据实际业务建立前后端开发通信协议并进行实现基于 SSM 整合技术开发任意业务模块功能 1 SpringMVC 简介 1.1 概述 1.1.1 web程序开发流程 【执行过程】…
阅读更多...
【搭环境】装Python3.8 open3d

【搭环境】装Python3.8 open3d

先装Python3.8 方法一试了找不到Python3.8的库&#xff0c;所以用方法二装上了。 Python3加入环境变量 更改Python默认指向 open3d需要Python3.6以上&#xff0c;最好用Ubuntu18版本&#xff0c;我用的16版本。。
阅读更多...
从入门到精通！Python数据分析畅销书《利用Python进行数据分析》第三版中文版助你成为数据分析师！

从入门到精通！Python数据分析畅销书《利用Python进行数据分析》第三版中文版助你成为数据分析师！

Python数据分析畅销书《利用Python进行数据分析》第三版中文版助你成为数据分析师&#xff01; 个人简介什么是数据分析如何自学数据分析书籍推荐作译者简介作者简介译者简介 主要变动导读视频&#xff1a;购书链接&#xff1a;参与方式往期赠书回顾 个人简介 &#x1f3d8;️&…
阅读更多...
【好玩的开源项目】Linux系统之部署proxx扫清黑洞小游戏

【好玩的开源项目】Linux系统之部署proxx扫清黑洞小游戏

【好玩的开源项目】Linux系统之部署proxx扫清黑洞小游戏 一、proxx小游戏介绍1.1 proxx小游戏简介1.2 开源地址 二、本地环境介绍2.1 本地环境规划2.2 本次实践介绍 三、检查本地环境3.1 检查本地操作系统版本3.2 检查系统内核版本 四、部署Node.js环境4.1 下载Node.js安装包4.…
阅读更多...
动态规划求二维网格中从左上角到右下角的最短路径( 每次只能向下、向右、向右下走 ) java 实现

动态规划求二维网格中从左上角到右下角的最短路径( 每次只能向下、向右、向右下走 ) java 实现

dp[i][j] 表示在以点(0,0)作为左上角&#xff0c;点(i,i) 作为右下角的二维网格中 左上角到右下角的最短路径&#xff0c; 动态转移方程为&#xff1a;dp[i][j] min{ dp[i][j-1],dp[i-1][j],dp[i-1][j-1] }.distance weight[i][j] ImageUtils.java&#xff1a; import java.a…
阅读更多...
vue3的 nextTick()的使用

vue3的 nextTick()的使用

引言&#xff1a; 当你修改了响应式状态时&#xff0c;DOM 会被自动更新。但是需要注意的是&#xff0c;DOM 更新不是同步的。Vue 会在“next tick”更新周期中缓冲所有状态的修改&#xff0c;以确保不管你进行了多少次状态修改&#xff0c;每个组件都只会被更新一次。 要等待…
阅读更多...
三十分钟学会Shell（上）

三十分钟学会Shell（上）

Shell ​ Shell 本身并不是内核的一部分&#xff0c;它只是站在内核的基础上编写的一个应用程序&#xff0c;是用户和Linux文件系统之间的桥梁。Shell 有自己的特殊性&#xff0c;就是开机立马启动&#xff0c;并呈现在用户面前&#xff1b;用户通过 Shell 来使用 Linux&#x…
阅读更多...
Windows日常故障自我排查：用工具eventvwr.msc(事件查看器）分析问题故障

Windows日常故障自我排查：用工具eventvwr.msc(事件查看器）分析问题故障

windows故障排查方法一&#xff1a; 工具用法 系统故障问题时&#xff0c;找不到解决方法 首先&#xff0c; 在搜索栏输入&#xff1a; 事件查看器(eventvwr.msc) 打开程序 根据程序找到程序运行的LOG 根据程序Operational筛选出错误日志&#xff1a; 日志中找错误原因&…
阅读更多...
假期宅家不无聊，用极米投影仪H6 Pro 4K解锁宅家多样玩法

假期宅家不无聊，用极米投影仪H6 Pro 4K解锁宅家多样玩法

不知道大家平时的假期是怎么度过的&#xff0c;是选择出门游玩?还是要在家好好安排自己的学习充电计划?平日工作这么累&#xff0c;不如趁假期好好休息一下&#xff0c;把以前落下的番剧补回来。不过手机的屏幕太小&#xff0c;电脑又不能躺在床上看&#xff0c;只有够大尺寸…
阅读更多...
如何解决EMI问题 电磁干扰问题诊断四步骤

如何解决EMI问题 电磁干扰问题诊断四步骤

电磁干扰的观念与防制﹐在国内已逐渐受到重视。虽然目前国内并无严格管制电子产品的电磁干扰&#xff08;EMI&#xff09;﹐但由于欧美各国多已实施电磁干扰的要求﹐加上数字产品的普遍使用﹐对电磁干扰的要求已是刻不容缓的事情。笔者由于啊作的关系﹐经常遇到许多产品已完成成…
阅读更多...
最新文章

Copyright @ 2022~2023