python之ORM
2024-09-01 21:10:58
pymysql
python操作数据库的基本步骤：
- 导入相应的python模块；
- 使用connect函数连接数据库，并返回一个connection对象；
- 通过connection对象的cursor方法，返回一个cursor对象；
- 通过cursor对象的execute方法执行SQL语句；
- 如果执行的是查询语句，通过cursor对象的fetchall语句获取返回结果；
- 调用cursor对象的close方法关闭cursor;
- 调用connection对象的close方法关闭数据库连接。
import pymysql conn = pymysql.Connect(host='127.0.0.1',user='admin',passwd='admin',db='test_py')
cur = conn.cursor() reCount = cur.execute('select * from student')
print(cur.fetchall()) # ((1, 'gareth', 22, datetime.date(2011, 9, 1)),) cur.close()
conn.close()
connection类成员
- begin: 开始事务
- commit： 提交事务
- rollback: 回滚事务
- cursor： 返回一个cursor对象
- autocommit： 设置是否事务自动提交
- set_character_set： 设置字符集编码
- get_server_info： 获取数据库版本信息
注： 一般不直接调用begin，commit和roolback函数，而是通过上下文管理器实现事务的提交与回滚操作。
cursor类成员对象：cursor对象表示数据库游标，用于执行SQL语句并获取SQL语句的执行结果。
- execute： 执行SQL语句
- close：关闭游标
- fetchall：获取SQL语句的所有记录
- fetchmany：获取SQL语句的多条记录
- fetchone：获取SQL语句的一条记录
- owncount：常量，表示SQL语句的结果集中返回了多少条记录
- arraysize：变量，保存了当前获取纪录的下标
- lastrowid：获取最新自增ID
注：在fetch数据时按照顺序进行，可以使用cursor.scroll(num,mode)来移动游标位置，如：
- cursor.scroll(1,mode='relative') # 相对当前位置移动
- cursor.scroll(2,mode='absolute') # 相对绝对位置移动
默认获取的数据是元祖类型，如果想要或者字典类型的数据，即：
conn = pymysql.Connect(host='127.0.0.1',user='admin',passwd='admin',db='test_py')
cur = conn.cursor(cursor=pymysql.cursors.DictCursor) reCount = cur.execute('select * from student')
print(cur.fetchall())
"""
[{'stu_id': 1, 'name': 'gareth', 'age': 22, 'register_data': datetime.date(2011, 9, 1)},
{'stu_id': 3, 'name': 'Bob', 'age': 19, 'register_data': datetime.date(2012, 2, 3)},
{'stu_id': 4, 'name': 'Bob', 'age': 19, 'register_data': datetime.date(2012, 2, 3)},
{'stu_id': 5, 'name': 'Mary', 'age': 18, 'register_data': datetime.date(2013, 1, 2)}]
""" cur.close()
conn.close()
使用上下文管理管理数据库:
import pymysql
import os def get_conn(**kwargs):
if os.getenv('DB','MYSQL') == 'MYSQL':
return pymysql.connect(host=kwargs.get('host','localhost'),
user=kwargs.get('user'),
passwd=kwargs.get('passwd'),
port=kwargs.get('port',3306),
db=kwargs.get('db')) def execute_sql(conn, sql):
with conn as cur:
cur.execute(sql) def insert_data(conn,sname,sage,sregister):
INSERT_FORMAT = """insert into student (name,age,register_data) values('{0}','{1}','{2}')"""
sql = INSERT_FORMAT.format(sname,sage,sregister)
execute_sql(conn,sql) def main():
conn = get_conn(host='127.0.0.1',
user='admin',
passwd='admin',
port=3306,
db='test_py') try:
insert_data(conn,'Bob',19,'2012-02-03')
insert_data(conn,'Mary',18,'2013-01-02') with conn as cur:
cur.execute('select * from student')
rows = cur.fetchall()
for row in rows:
print(row)
finally:
if conn:
conn.close() if __name__ == '__main__':
main()
上面例子中如果values('{0}','{1}','{2}')的引号去掉，则会报错：pymysql.err.InternalError: (1054, "Unknown column 'jack' in 'field list'")
cur.execute("insert into student (name,age,register_data) values('jack',12,'2012-02-03')")
INSERT_FORMAT = """insert into student (name,age,register_data) values('{0}','{1}','{2}')"""
批量插入：
cur = conn.cursor()
cur.executemany("insert into student (name,age,register_data) values(%s,%s,%s)",
[('jack',12,'2012-02-03'),('Bob',12,'2012-02-03')] )
ORM
orm英文全称object relational mapping，即对象映射关系程序，简单来说我们类似python这种面向对象的程序来说一切皆对象，但是我们使用的数据库却都是关系型的，为了保证一致的使用习惯，通过orm将编程语言的对象模型和数据库的关系模型建立映射关系，这样我们在使用编程语言对数据库进行操作的时候可以直接使用编程语言的对象模型进行操作就可以了，而不用直接使用sql语言。
优点：
- 隐藏了数据访问细节，“封闭”的通用数据库交互，ORM的核心。他使得我们的通用数据库交互变得简单易行，并且完全不用考虑该死的SQL语句。快速开发，由此而来。
- ORM使我们构造固化数据结构变得简单易行。
缺点：
- 自动化意味着映射和关联管理，代价是牺牲性能
sqlalchemy
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
SQLAlchemy本身无法操作数据库，其必须依赖pymsql等第三方插件，Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作，如：
MySQL-Python
mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname> pymysql
mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>] MySQL-Connector
mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname> cx_Oracle
oracle+cx_oracle://user:pass@host:port/dbname[?key=value&key=value...] 更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html
dialect
使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。
from sqlalchemy import create_engine
engine = create_engine('mysql+pymysql://user_account:password@127.0.0.1/test', max_overflow=5)
cur = engine.execute("insert into student(name,age,register_date) values('Jack',20,'2018-01-02')")
print(cur.lastrowid)
#
cur = engine.execute("insert into student(name,age,register_date) values(%s,%s,%s)",
[('Gareth',18,'2018-07-02'),('Mar',17,'2017-12-02')])
cur = engine.execute("insert into student(name,age,register_date) values(%(name)s,%(age)s,%(register_date)s)",
name = 'Ker',age = 21,register_date='2016-09-01')
cur = engine.execute('select * from student')
print(cur.fetchone())
# (1, 'Jack', 20, datetime.date(2018, 1, 2))
print(cur.fetchmany(2))
# [(2, 'Gareth', 18, datetime.date(2018, 7, 2)), (3, 'Mar', 17, datetime.date(2017, 12, 2))]
print(cur.fetchall())
# [(4, 'Ker', 21, datetime.date(2016, 9, 1))]
ORM使用
使用 Engine/ConnectionPooling/Dialect 进行数据库操作，Engine使用ConnectionPooling连接数据库，然后再通过Dialect执行SQL语句。
创建表：
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,ForeignKey, UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://user_account:password@127.0.0.1:3306/test", max_overflow = 5)
Base = declarative_base() # 生成一个基类 class Classes(Base):
__tablename__='classes'
id = Column(Integer, primary_key=True)
name = Column(String(32)) Base.metadata.create_all(engine) #创建表结构
创建与数据库的会话session:
session_class = sessionmaker(bind=engine) # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
session = session_class() # 生成session实例 cla_obj = Classes(name='python099') #生成你要创建的数据对象 session.add(cla_obj) # 把要创建的数据对象添加到这个session里， 一会统一创建 session.commit() # 现此才统一提交，创建数据
查询
sqlalchemy把返回的数据映射成一个对象，调用每个字段可以像调用对象属性一样。
session_class = sessionmaker(bind=engine) # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
session = session_class() # 生成session实例 my_classes = session.query(Classes)
print(my_classes)
# "SELECT classes.id AS classes_id, classes.name AS classes_name FROM classes" my_classes = session.query(Classes).first()
print(my_classes) # 返回内存对象地址
# <__main__.Classes object at 0x000001832CE1A4E0> my_classes = session.query(Classes).first()
print('id: ',my_classes.id,'name:',my_classes.name)
# id: 1 name: python011 my_classes = session.query(Classes)
for classes in my_classes:
# print('id: ',classes.id,'name:',classes.name)
# id: 1 name: python011
# id: 2 name: python016
# id: 3 name: linux
把内存对象地址转化为可读性数据，除了调用字段还可以使用__repr__(self) 函数。
def __repr__(self):
return "Classes(id=%s,name=%s)"%(self.id, self.name) my_classes = session.query(Classes)
print(my_classes) # SELECT classes.id AS classes_id, classes.name AS classes_name FROM classes print(my_classes[1]) # Classes(id=2,name=python016) for classes in my_classes:
print(classes)
# Classes(id=1,name=python011)
# Classes(id=2,name=python016)
# Classes(id=3,name=linux)
过滤
print(session.query(Classes).filter(Classes.name.in_(['mysql','python_sqlalchemy'])).all()) session.query(Classes).filter_by(name='python011').first()
多条件查询： 相当于id > 1 and id <4
obj = session.query(Classes).filter(Classes.id>1).filter(Classes.id<4).all()
print(obj)
# [Classes(id=2,name=python016), Classes(id=3,name=linux)]
获取所有数据
print(session.query(Classes).all())
# [Classes(id=1,name=python_mysql), Classes(id=2,name=python016), Classes(id=3,name=linux)]
print(session.query(Classes.id,Classes.name).all())
# [(1, 'python_mysql'), (2, 'python016'), (3, 'linux')]
查询语法
Common Filter Operators
Here’s a rundown of some of the most common operators used in filter():
equals:
query.filter(User.name == 'ed')
not equals:
query.filter(User.name != 'ed')
LIKE:
query.filter(User.name.like('%ed%'))
IN:
NOT IN:
query.filter(~User.name.in_(['ed', 'wendy', 'jack']))
IS NULL:
IS NOT NULL:
AND:
2.1. ObjectRelationalTutorial 17
query.filter(User.name.in_(['ed', 'wendy', 'jack']))
# works with query objects too:
query.filter(User.name.in_( session.query(User.name).filter(User.name.like('%ed%'))
))
query.filter(User.name == None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.is_(None))
query.filter(User.name != None)
# alternatively, if pep8/linters are a concern
query.filter(User.name.isnot(None))
SQLAlchemy Documentation, Release 1.1.0b1
# use and_()
from sqlalchemy import and_
query.filter(and_(User.name == 'ed', User.fullname == 'Ed Jones'))
# or send multiple expressions to .filter()
query.filter(User.name == 'ed', User.fullname == 'Ed Jones')
# or chain multiple filter()/filter_by() calls
query.filter(User.name == 'ed').filter(User.fullname == 'Ed Jones')
Note: Makesureyouuseand_()andnotthePythonandoperator! • OR:
Note: Makesureyouuseor_()andnotthePythonoroperator! • MATCH:
query.filter(User.name.match('wendy'))
Note: match() uses a database-specific MATCH or CONTAINS f
分组和统计
print(session.query(Classes).filter(Classes.name.like('p%')).count())
#
print(session.query(Classes).count())
#
from sqlalchemy import func
print(session.query(Classes).all())
# [Classes(id=1,name=python_mysql), Classes(id=2,name=python016), Classes(id=3,name=linux), Classes(id=5,name=linux)]
print(session.query(func.count(Classes.name),Classes.name).group_by(Classes.name).all())
#[(1, 'python_mysql'), (1, 'python016'), (2, 'linux')]
修改
使用修改，可以加个判断，如果没有查询到需要修改的信息则跳过，否则容易出现异常 AttributeError 。
my_classes = session.query(Classes).filter_by(name='python011').first() my_classes.name = 'python_mysql022'
session.commit()
my_classes = session.query(Classes)
for classes in my_classes:
print(classes) # Classes(id=1,name=python_mysql)
# Classes(id=2,name=python016)
# Classes(id=3,name=linux)
异常：
Traceback (most recent call last):
File "C:/D/personal_data/workspace/eleven/mysql_study/class_study/orm_01.py", line 64, in <module>
my_classes.name = 'python_mysql022'
AttributeError: 'NoneType' object has no attribute 'name'
回滚
my_classes = session.query(Classes).filter_by(id=2).first()
my_classes.name = 'python_sqlalchemy' fake_classes = Classes(name='mysql')
session.add(fake_classes) print(session.query(Classes).filter(Classes.name.in_(['mysql','python_sqlalchemy'])).all())
#[Classes(id=2,name=python_sqlalchemy), Classes(id=4,name=mysql)]
session.rollback() print(session.query(Classes).filter(Classes.name.in_(['mysql','python_sqlalchemy'])).all())
# []
程序：
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String,ForeignKey, UniqueConstraint,Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine engine = create_engine("mysql+pymysql://jiawenyx:intel@3117@127.0.0.1:3306/test", max_overflow = 5)
Base = declarative_base() class Classes(Base):
__tablename__='classes'
id = Column(Integer, primary_key=True)
name = Column(String(32)) def __repr__(self):
return "Classes(id=%s,name=%s)"%(self.id, self.name) Base.metadata.create_all(engine) #创建表结构 session_class = sessionmaker(bind=engine) # 创建与数据库的会话session class ,注意,这里返回给session的是个class,不是实例
session = session_class() # 生成session实例
#
# cla_obj = Classes(name='python016') #生成你要创建的数据对象
# cla_obj2 = Classes(name='linux')
#
# session.add(cla_obj) # 把要创建的数据对象添加到这个session里， 一会统一创建
# session.add(cla_obj2)
# session.commit() # 现此才统一提交，创建数据 # ******************************** 查询 *************************************** my_classes = session.query(Classes)
# print(my_classes)
# "SELECT classes.id AS classes_id, classes.name AS classes_name FROM classes"
my_classes = session.query(Classes).first()
# print(my_classes)
# <__main__.Classes object at 0x000001832CE1A4E0>
my_classes = session.query(Classes).first()
# print('id: ',my_classes.id,'name:',my_classes.name)
# id: 1 name: python011 my_classes = session.query(Classes)
# for classes in my_classes:
# print('id: ',classes.id,'name:',classes.name)
# id: 1 name: python011
# id: 2 name: python016
# id: 3 name: linux # my_classes = session.query(Classes)
# print(my_classes) # SELECT classes.id AS classes_id, classes.name AS classes_name FROM classes
# print(my_classes[1]) # Classes(id=2,name=python016)
# for classes in my_classes:
# print(classes)
# # Classes(id=1,name=python011)
# # Classes(id=2,name=python016)
# # Classes(id=3,name=linux) #****************************************** 修改 ******************************************** my_classes = session.query(Classes).filter_by(name='python011').first() # my_classes.name = 'python_mysql022'
# session.commit()
# my_classes = session.query(Classes)
# for classes in my_classes:
# print(classes) # Classes(id=1,name=python_mysql)
# Classes(id=2,name=python016)
# Classes(id=3,name=linux) # ******************************* 回滚 ******************************** my_classes = session.query(Classes).filter_by(id=2).first()
my_classes.name = 'python_sqlalchemy' fake_classes = Classes(name='mysql')
session.add(fake_classes) # print(session.query(Classes).filter(Classes.name.in_(['mysql','python_sqlalchemy'])).all())
#[Classes(id=2,name=python_sqlalchemy), Classes(id=4,name=mysql)]
session.rollback() # print(session.query(Classes).filter(Classes.name.in_(['mysql','python_sqlalchemy'])).all()) # **********************获取所有数据***************************** # print(session.query(Classes).all())
# # [Classes(id=1,name=python_mysql), Classes(id=2,name=python016), Classes(id=3,name=linux)]
# print(session.query(Classes.id,Classes.name).all())
# # [(1, 'python_mysql'), (2, 'python016'), (3, 'linux')]
#
# # ********************** 多条件查询 ****************************
#
# obj = session.query(Classes).filter(Classes.id>1).filter(Classes.id<4).all()
# print(obj)
# # [Classes(id=2,name=python016), Classes(id=3,name=linux)]
#
# # ********************** 统计和分组 ****************************
#
print(session.query(Classes).filter(Classes.name.like('p%')).count())
#
print(session.query(Classes).count())
# from sqlalchemy import func
print(session.query(Classes).all())
# [Classes(id=1,name=python_mysql), Classes(id=2,name=python016), Classes(id=3,name=linux), Classes(id=5,name=linux)]
print(session.query(func.count(Classes.name),Classes.name).group_by(Classes.name).all())
#[(1, 'python_mysql'), (1, 'python016'), (2, 'linux')]
程序
外键
from sqlalchemy import Column,Integer,String
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker engine = create_engine("mysql+pymysql://user:password@127.0.0.1:3306/test", max_overflow = 5) Base = declarative_base() class Customer(Base):
__tablename__='customer'
id = Column(Integer,primary_key=True)
name = Column(String(32))
billing_address_id = Column(Integer,ForeignKey('address.id'))
shipping_address_id = Column(Integer)
#shipping_address_id = Column(Integer,ForeignKey('address.id'))
address = relationship('Address', backref='customer')
# shipping_address = relationship('Address')
def __repr__(self):
return "id:%s,name:%s,billing_address_id:%s,shipping_address_id:%s"\
%(self.id,self.name,self.billing_address_id,self.shipping_address_id) class Address(Base):
__tablename__ = 'address'
id = Column(Integer,primary_key=True)
stress = Column(String(64))
city = Column(String(64))
state = Column(String(64)) def __repr__(self):
return 'id:%s,stress:%s,city:%s,state:%s' %(self.id,self.stress,self.city,self.state) Base.metadata.create_all(engine) #创建表结构 session_class = sessionmaker(engine)
session = session_class() # cus1 = Customer(name='gareth',billing_address_id=1, shipping_address_id = 2)
# cus2 = Customer(name = 'Jack',billing_address_id=2, shipping_address_id = 3)
# cus3 = Customer(name ='Mar',billing_address_id=3, shipping_address_id = 3)
#
# add_obj = Address(stress = 'shuangxing', city = 'shunyi', state= 'Beijing')
# add_obj1 = Address(stress = 'tiantongyuan', city = 'changping', state = 'Beijing')
# add_obj2 = Address(stress = 'bayiqiao', city = 'nanchang', state = 'jiangxi')
# session.add_all([add_obj,add_obj1,add_obj2,cus1,cus2,cus3])
#
# session.commit() obj = session.query(Address).first()
# print(obj) # id:1,stress:shuangxing,city:shunyi,state:Beijing
# print(obj.customer) # [id:1,name:gareth,billing_address_id:1,shipping_address_id:2]
# for i in obj.customer:
# print(i) # id:1,name:gareth,billing_address_id:1,shipping_address_id:2 obj = session.query(Customer).first()
# print(obj) # id:1,name:gareth,billing_address_id:1,shipping_address_id:2
# print(obj.address) # id:1,stress:shuangxing,city:shunyi,state:Beijing
外键
注：
定义外键： billing_address_id = Column(Integer,ForeignKey('address.id'))
relationship：address = relationship('Address', backref='customer')，通过Address表可以反查customer信息，可以在Customer表中通过address字段，查找Adress表中信息；从而两张表互查。
obj = session.query(Customer).first()
print(obj) # id:1,name:gareth,billing_address_id:1,shipping_address_id:2
print(obj.address) # id:1,stress:shuangxing,city:shunyi,state:Beijing
obj = session.query(Address).first()
print(obj) # id:1,stress:shuangxing,city:shunyi,state:Beijing
print(obj.customer) # [id:1,name:gareth,billing_address_id:1,shipping_address_id:2]
for i in obj.customer:
print(i) # id:1,name:gareth,billing_address_id:1,shipping_address_id:2
多外键关联
from sqlalchemy import Column,Integer,String
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker engine = create_engine("mysql+pymysql://user:password@127.0.0.1:3306/test", max_overflow = 5) Base = declarative_base() class Customer(Base):
__tablename__='customer'
id = Column(Integer,primary_key=True)
name = Column(String(32))
billing_address_id = Column(Integer,ForeignKey('address.id'))
shipping_address_id = Column(Integer,ForeignKey('address.id'))
address = relationship('Address')
shipping_address = relationship('Address')
def __repr__(self):
return "id:%s,name:%s,billing_address_id:%s,shipping_address_id:%s"\
%(self.id,self.name,self.billing_address_id,self.shipping_address_id) class Address(Base):
__tablename__ = 'address'
id = Column(Integer,primary_key=True)
stress = Column(String(64))
city = Column(String(64))
state = Column(String(64)) def __repr__(self):
return 'id:%s,stress:%s,city:%s,state:%s' %(self.id,self.stress,self.city,self.state) Base.metadata.create_all(engine) #创建表结构 session_class = sessionmaker(engine)
session = session_class()
创建表结构OK，但Address表中插入数据时会报下面的错
sqlalchemy.exc.AmbiguousForeignKeysError: Could not determine join condition between parent/child tables on relationship Customer.address -
there are multiple foreign key paths linking the tables. Specify the 'foreign_keys' argument, providing a list of those columns which should be counted as containing a foreign key reference to the parent table.
修改方法：
class Customer(Base):
__tablename__='customer'
id = Column(Integer,primary_key=True)
name = Column(String(32))
billing_address_id = Column(Integer,ForeignKey('address.id'))
shipping_address_id = Column(Integer,ForeignKey('address.id'))
billing_address = relationship('Address',foreign_keys=[billing_address_id])
shipping_address = relationship('Address',foreign_keys=[shipping_address_id])
处理中文
sqlalchemy设置编码字符集一定要在数据库访问的URL上增加charset=utf8，否则数据库的连接就不是utf8的编码格式
eng = create_engine('mysql://root:root@localhost:3306/test2?charset=utf8',echo=True)
多对多关系
ORM_API
from sqlalchemy import Column,Integer,String,Date,Table
from sqlalchemy import ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker engine = create_engine("mysql+pymysql://admin:admin@localhost:3306/test_py?charset=utf8", max_overflow = 5) Base = declarative_base() book_m2m_author = Table('book_m2m_author', Base.metadata,
Column('book_id',Integer,ForeignKey('books.id')),
Column('author_id',Integer,ForeignKey('authors.id')),
)
class Book(Base):
__tablename__='books'
id = Column(Integer,primary_key=True)
name = Column(String(64))
pub_data = Column(Date)
authors = relationship('Author',secondary=book_m2m_author, backref='books') def __repr__(self):
return self.name class Author(Base):
__tablename__='authors'
id = Column(Integer,primary_key=True)
name = Column(String(32)) def __repr__(self):
return self.name Base.metadata.create_all(engine) #创建表结构
session:
from sqlalchemy.orm import sessionmaker
import book_orm Session = sessionmaker(book_orm.engine)
session = Session() b1 = book_orm.Book(name='三国演义')
b2 = book_orm.Book(name='水浒传')
b3 = book_orm.Book(name='红楼梦')
b4 = book_orm.Book(name='西游记') a1 = book_orm.Author(name='罗贯中')
a2= book_orm.Author(name='曹雪芹')
a3= book_orm.Author(name='吴承恩')
a4= book_orm.Author(name='施耐庵')
a5= book_orm.Author(name='Gareth')
a6= book_orm.Author(name='Yu')
a7= book_orm.Author(name='caven') b1.authors = [a1,a6]
b2.authors = [a4,a5,a7]
b3.authors = [a2]
b4.authors = [a3] session.add_all([a1,a2,a3,a4,a5,a6,a7,b1,b2,b3,b4])
session.commit()
table:
mysql> show tables;
+-------------------+
| Tables_in_test_py |
+-------------------+
| authors |
| book_m2m_author |
| books |
| classes |
| student |
| user |
+-------------------+
6 rows in set (0.03 sec) mysql> select * from authors;
+----+--------+
| id | name |
+----+--------+
| 1 | 罗贯中 |
| 2 | Yu |
| 3 | 曹雪芹 |
| 4 | 吴承恩 |
| 5 | 施耐庵 |
| 6 | Gareth |
| 7 | caven |
+----+--------+
7 rows in set (0.03 sec) mysql> select * from authors;
+----+--------+
| id | name |
+----+--------+
| 1 | 罗贯中 |
| 2 | Yu |
| 3 | 曹雪芹 |
| 4 | 吴承恩 |
| 5 | 施耐庵 |
| 6 | Gareth |
| 7 | caven |
+----+--------+
7 rows in set (0.03 sec) mysql> select * from book_m2m_author;
+---------+-----------+
| book_id | author_id |
+---------+-----------+
| 4 | 5 |
| 4 | 6 |
| 1 | 1 |
| 1 | 2 |
| 2 | 3 |
| 4 | 7 |
| 3 | 4 |
+---------+-----------+
7 rows in set (0.00 sec)
tables
查询：
print("通过book table查询关联的作者")
book_obj = session.query(book_orm.Book).filter_by(name="三国演义").all()
print(book_obj) # [三国演义]
book_obj = session.query(book_orm.Book).filter_by(name="三国演义").first()
print(book_obj.name, book_obj.authors)
#三国演义 [罗贯中, 曹雪芹]
print("通过author table查询关联的书")
author_obj = session.query(book_orm.Author).filter_by(name="曹雪芹").first()
print(author_obj.name, author_obj.books)
直接删除作者：
author_obj = session.query(book_orm.Author).filter(book_orm.Author.id>7).all()
print(author_obj) for i in author_obj:
session.delete(i)
session.commit()
print(session.query(book_orm.Author).all()) #[罗贯中, 曹雪芹, 施耐庵, Yu, 罗贯中, 曹雪芹, 施耐庵, Yu, 罗贯中, 曹雪芹, 施耐庵, Yu, 罗贯中, 曹雪芹, 施耐庵, Yu, 罗贯中, 罗贯中, 曹雪芹, 罗贯中, 曹雪芹]
#[罗贯中, Yu, 曹雪芹, 吴承恩, 施耐庵, Gareth, caven]
通过书删除作者：
author_obj = session.query(book_orm.Author).filter_by(name="曹雪芹").first()
book_obj = session.query(book_orm.Book).filter_by(name="红楼梦").first()
book_obj.authors.remove(author_obj)
session.commit()
删除数据时不用管boo_m2m_authors ， sqlalchemy会自动帮你把对应的数据删除。
最新文章
- title换行
- Revolving Digits（hdu 4333）
- 使用Runtime.getRuntime().exec()在java中调用python脚本
- AngularJS服务
- Atitit. 解压缩zip文件 的实现最佳实践 java c# .net php
- ZooKeeper学习第六期---ZooKeeper机制架构
- select @@identity的用法
- Ajax清除浏览器js、css、图片缓存的方法
- 【原】lua的table深拷贝
- sprintf函数php的详细使用方法
- css3 文本超出后出现省略号
- Android 使用BroadcastReceiver的几种方法
- 混合使用Azure LB和ILB访问相同web服务（2）
- Android自定义照相机实现（拍照、保存到SD卡，利用Bundle在Acitivity交换数据）
- ioctl、文件操作接口函数以及nand的升级模式的操作过程详解
- 为HttpStatusCodeResult加入customErrors
- ztree异步加载
- sprintf格式化字符串带来的注入隐患
- c# 操作Word总结（车）
- 音视频处理ffmpeg使用