sqlalchemy示例代码
多对多外键
场景：一个作者映射多个书籍，一个书籍有多个作者 作者表： id name email sex 1 Alex alex@.. M 2 Rail rail@.. W 书籍表： id name 1 book1 2 book2 作者书籍映射关系表： id book_id anthor_id 1 1 1 2 1 2 3 2 1 【Alex同时参与了两本书的创作】 【book1 有两个作者参与创作】
#!/usr/bin/env python # Author:zhangmingda '''''' from sqlalchemy import create_engine,ForeignKey,DATE,Table from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship,sessionmaker engine = create_engine("mysql+pymysql://zhangmingda:Wyf@1314@120.92.133.227/oldboy_db?charset=utf8",encoding='utf8',echo=False) '''说明：engine 如果想写入中文，在库名后面加入?charset=utf8 encoding=utf-8 不起作用''' '''生成基类''' 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' #表的名字 __table_args__ = {'mysql_charset': 'utf8'} id = Column(Integer,primary_key=True) name = Column(String(64)) pub_date = Column(DATE) authors = relationship('Author',secondary=book_m2m_author,backref='books') #如何对应第三张表 def __repr__(self): return self.name class Author(Base): __tablename__ = 'authors' __table_args__ = {'mysql_charset': 'utf8'} #创建表的字符编码设置 id = Column(Integer,primary_key=True) name = Column(String(32)) # books = relationship('Book',secondary=book_m2m_author,backref='authors')#如何对应第三张表，作者表或书籍表有一个写了即可 def __repr__(self): return self.name Base.metadata.create_all(engine)
#!/usr/bin/env python # Author:zhangmingda import orm_many_to_many_foreign_key as m2m from sqlalchemy.orm import sessionmaker SessionClass = sessionmaker(bind=m2m.engine) session = SessionClass() '''作者数据''' au1 = m2m.Author(name='Alex') au2 = m2m.Author(name='Rain') au3 = m2m.Author(name='Jack') '''书籍数据''' b1 = m2m.Book(name='跟Alex学Python',pub_date ='2018-09-17' ) b2 = m2m.Book(name='跟Alex学装逼',pub_date ='2018-09-20' ) b3 = m2m.Book(name='跟Alex学把妹',pub_date ='2018-09-22' ) b4 = m2m.Book(name='跟Alex学把妹',pub_date ='2018-09-22' ) '''映射关系''' b1.authors = [au1,au2,au3] b2.authors = [au1,au3] b3.authors = [au2,au3] # session.add_all([au1,au2,au3,b1,b2,b3])#创建数据 session.commit() '''============查询:通过书看作者都有谁/通过作者看都创建了啥书====================''' b1 = session.query(m2m.Book).filter_by(name='跟Alex学Python').first() print(b1,'的作者是：',b1.authors) a1 = session.query(m2m.Author).filter_by(name='Alex').first() print(a1,'创作的书籍有：',a1.books) '''====删除：多对多删除数据时不用管book_m2m_author ， sqlalchemy会自动帮你把对应的数据删除 通过书删除作者''' b3 = session.query(m2m.Book).filter_by(name='跟Alex学装逼').first() print(b3,'的作者有：',b3.authors) print('删除作者Jack') jack = session.query(m2m.Author).filter_by(name='Jack').first() b3.authors.remove(jack) print(b3,'创作的书籍有：',b3.authors) session.commit()#提交更改。 '''==============多对多增加数据===========''' b3 = session.query(m2m.Book).filter_by(name='跟Alex学把妹').first() print('增加之前，',b3.name,'作者有：',b3.authors) author = session.query(m2m.Author).filter_by(name='Alex').first() #找到作者 b3.authors.append(author) #向书籍中append作者 print('增加作者',author.name) print(b3,'的作者有：',b3.authors) session.commit() '''增删改查，第三张表会自动跟着修改。无需多管'''
多对一外键
#!/usr/bin/env python # Author:zhangmingda '''一张表中有两个字段同时外键到另外一个表''' from sqlalchemy import create_engine,ForeignKey,DATE from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship,sessionmaker engine = create_engine("mysql+pymysql://zhangmingda:Wyf@1314@120.92.133.227/oldboy_db?charset=utf8",encoding='utf8',echo=False) '''生成基类''' Base = declarative_base() class Customer(Base): '''顾客''' __tablename__ = 'customer' #要创建的表名 __table_args__ = {'mysql_charset':'utf8'} id = Column(Integer,primary_key=True) name = Column(String(32),nullable=False) billing_address_id = Column(Integer,ForeignKey('addresses.id')) shipping_address_id = Column(Integer,ForeignKey('addresses.id')) '''下面的foregin_keys 指的是通过哪个外键字段去找数据。''' billing_address = relationship("Addresses",foreign_keys=[billing_address_id]) shipping_address = relationship("Addresses",foreign_keys=[shipping_address_id]) class Addresses(Base): __tablename__ = 'addresses' __table_args__ = {'mysql_charset': 'utf8'} id = Column(Integer,primary_key=True) street = Column(String(64)) city = Column(String(64)) state = Column(String(64)) def __repr__(self): '''被关联的数据正向查询时，return内容作为返回值''' return "Address is: state:%s city:%s street:%s " % (self.state,self.city,self.street) # Base.metadata.create_all(engine) # 创建表结构
#!/usr/bin/env python # Author:zhangmingda from day12.多对一外键到同一张表 import foreginkey_many_orm from sqlalchemy.orm import sessionmaker Session_class = sessionmaker(bind=foreginkey_many_orm.engine) #生成数据库连接类(这里的返回值是个类) session = Session_class() #生成Session实例 '''为addresses表准备数据''' addr1 = foreginkey_many_orm.Addresses(street='Zhuxinzhuang', city="ChangPing", state="BJ") addr2 = foreginkey_many_orm.Addresses(street='Wudaokou', city="Haidian", state="BJ") addr3 = foreginkey_many_orm.Addresses(street='YanJiao', city="Langfang", state="HB") # session.add_all([addr1,addr2,addr3]) #创建数据 '''为Customer表准备人员数据''' c1 = foreginkey_many_orm.Customer(name='Alex', billing_address=addr1, shipping_address=addr2) c2 = foreginkey_many_orm.Customer(name='Rain', billing_address=addr3, shipping_address=addr3) c3 = foreginkey_many_orm.Customer(name='K', billing_address_id=1, shipping_address_id=2) c4 = foreginkey_many_orm.Customer(name='Jack', billing_address_id=3, shipping_address_id=3) # session.add_all([c1,c2,c3,c4]) #c1,c2 和c3,c4两种方式的写法都行，实际应用可能为前者#创建数据 obj = session.query(foreginkey_many_orm.Customer).filter_by(name='Jack').first() print(obj.name,'billing_address:',obj.billing_address,'\n','shipping_address:',obj.shipping_address) session.commit()
sqlalchemy
#!/usr/bin/env python # Author:zhangmingda '''数据的增删改查''' from sqlalchemy import create_engine #用来建立连接 from sqlalchemy.ext.declarative import declarative_base #创建表的基类 from sqlalchemy import Column,Integer,String # 表结构数据类型 from sqlalchemy.orm import sessionmaker '''建立连接''' engine = create_engine("mysql+pymysql://zhangmingda:Wyf@1314@120.92.133.227/oldboy_db",encoding='utf-8',echo=False) '''生成基类''' Base = declarative_base() '''子类''' class sub_base(Base): __tablename__ = 'user_talbe' #要创建的表名 id = Column(Integer,primary_key=True) username = Column(String(32)) password = Column(String(64)) '''创建表''' Base.metadata.create_all(engine) #创建表结构 '''插入数据之 创建session实例''' Session_class = sessionmaker(bind=engine) #生成数据库连接类(这里的返回值是个类) Session = Session_class() #生成Session实例 '''准备数据对象''' data_obj = sub_base(username='Qjj',password='test14413432')#生成创建的数据对象 # Session.add(data_obj) #将数据写入到连接的实例里面 print(data_obj.username,data_obj.id) #还没创建 Session.commit() #提交创建数据 # print(data_obj.username,data_obj.id) '''查询数据''' users = Session.query(sub_base).filter_by(username='Qjj') print(users,'type users',type(users)) #返回所有数据对象 for user in users: print(user.id,user.username,user.password) user_one = Session.query(sub_base).filter_by(username='zhangmingda').first() print(user_one.id,user_one.username,user_one.password) '''修改数据''' user_one.username = 'ZHANGMINGDA' Session.commit() '''回滚测试''' my_user = Session.query(sub_base).filter_by(id=1).first() my_user.username = 'alex' new_user = sub_base(username='ALEX',password='12345') Session.add(new_user) print('回滚前',Session.query(sub_base).filter(sub_base.username.in_(['LAEX','alex'])).all()) Session.rollback() print('回滚后',Session.query(sub_base).filter(sub_base.username.in_(['LAEX','alex'])).all()) Session.commit() '''获取所有数据(每个数据作为元组，多个数据组成元组为元素的列表)''' print(Session.query(sub_base.id,sub_base.username,sub_base.password).all()) '''多条件查询''' objs = Session.query(sub_base).filter(sub_base.id>0).filter(sub_base.id<7).all() print('多条件查询结果') for obj in objs: print(obj.id,obj.username,obj.password) '''统计个数''' count = Session.query(sub_base).filter(sub_base.username.like('Zhangmingda')).count() print('统计符合条件的个数：',count) '''分组查询''' from sqlalchemy import func print('分组查询统计：',Session.query(sub_base.username,func.count(sub_base.username)).group_by(sub_base.username).all()) '''注意这里query 里面的结构：写的是查询出来的结果显示顺序，本例(sub_base.username,func.count(sub_base.username) 为用户名和统计的数字''' '''删除数据''' obj_del = Session.query(sub_base).filter(sub_base.id>5).filter(sub_base.username=='Qjj').all() for d_obj in obj_del: Session.delete(d_obj) Session.commit()
#!/usr/bin/env python # Author:zhangmingda '''外键两张表的正反向查询''' from sqlalchemy import create_engine,ForeignKey,DATE from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship,sessionmaker engine = create_engine("mysql+pymysql://zhangmingda:Wyf@1314@120.92.133.227/oldboy_db?charset=utf8",encoding='utf8',echo=False) '''生成基类''' Base = declarative_base() class Students(Base): '''学生类''' __tablename__ = 'students' #要创建的表名 __table_args__ = {'mysql_charset':'utf8'} id = Column(Integer,primary_key=True) name = Column(String(32),nullable=False) register_date = Column(DATE,nullable=False) def __repr__(self): return "<%s name:%s>"%(self.id,self.name) class Students_Record(Base): __tablename__ = 'stu_record' __table_args__ = {'mysql_charset': 'utf8'} id = Column(Integer,primary_key=True) day = Column(Integer,nullable=False) status = Column(String(32),nullable=False) stu_id = Column(Integer,ForeignKey("students.id")) student = relationship('Students',backref = 'record') def __repr__(self): '''被关联的数据正向查询时，return内容作为返回值''' return "record_id:%s %s: day:%s status:%s" % (self.id,self.student.name,self.day,self.status) Base.metadata.create_all(engine) # 创建表结构 Session_class = sessionmaker(bind=engine) #生成数据库连接类(这里的返回值是个类) session = Session_class() #生成Session实例 stu1 = Students(name='Alex',register_date="2018-09-16") stu2 = Students(name='Jack',register_date='2018-09-17') stu3 = Students(name='Rain',register_date='2018-09-18') stu4 = Students(name='Eric',register_date='2018-09-19') stu5 = Students(name='Qjj',register_date='2018-09-10') record1 = Students_Record(day=1,status='YES',stu_id=1) record2 = Students_Record(day=2,status='YES',stu_id=1) record3 = Students_Record(day=3,status='No',stu_id=1) record4 = Students_Record(day=4,status='YES',stu_id=2) record5 = Students_Record(day=1,status='YES',stu_id=3) # session.add_all([record1,record2,record3,record4,record5]) session.commit() s1 = session.query(Students).first() #查询符合条件的单条数据 print(s1) print(s1.record)#通过外键关系反向调第二张表里面和这个数据有关的所有数据，作为列表返回 for record in s1.record: print(record) record_status = session.query(Students_Record).filter_by(status='YES').all() #找所有没上课的记录，返回列表 for r in record_status: #循环列表，通过上课记录的表没上课的记录反向找学生表里面这个人是谁 print('%s day %s 没上课'% (r.student.name,r.day)) s2 = session.query(Students).all() #查询符合条件的所有数据 # print(s2) # print(s2.record)#通过外键关系反向调第二张表里面和这个数据有关的所有数据，作为列表返回 for s in s2: for record in s.record: print(record)
#!/usr/bin/env python # Author:zhangmingda import sqlalchemy '''表的创建 查询''' from sqlalchemy import create_engine,ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy import Column, Integer, String from sqlalchemy.orm import relationship,sessionmaker engine = create_engine("mysql+pymysql://zhangmingda:Wyf@1314@120.92.133.227/oldboy_db",encoding='utf-8',echo=False) '''生成基类''' Base = declarative_base() class sub_base(Base): __tablename__ = 'user_talbe' #要创建的表名 id = Column(Integer,primary_key=True) username = Column(String(32)) password = Column(String(64)) class Address(Base): __tablename__ = 'addresses' id = Column(Integer,primary_key=True) email_address = Column(String(32),nullable=False) user_id = Column(Integer,ForeignKey('user_talbe.id')) user = relationship('sub_base',backref='addresses') def __repr__(self): '''被关联的数据正向查询时，return内容作为返回值''' return "Address(email_address='%s')" % self.email_address Base.metadata.create_all(engine) # 创建表结构 Session_class = sessionmaker(bind=engine) #生成数据库连接类(这里的返回值是个类) Session = Session_class() #生成Session实例 '''在user_talbe表中查询addresses表中的数据''' obj = Session.query(sub_base).first() for address in obj.addresses: print(address) '''反向查询''' add_obj = Session.query(Address).first() print(add_obj.user.username) '''连表查询''' # ret = Session.query(sub_base,Address).filter(sub_base.id == Address.id).all() # ret = Session.query(sub_base).join(Address).all() ret = Session.query(sub_base).join(Address,isouter=True).all() # print(ret) for obj in ret: # print(dir(obj)) print(obj.addresses,obj.username,obj.password,obj.id)
pymysql read wirte 查询 & 创建数据
#!/usr/bin/env python # Author:zhangmingda import pymysql #读取数据库中的数据 conn = pymysql.connect(host='120.92.133.227',port=3306,user='zhangmingda',passwd='Wyf@1314',db='oldboy_db') # cursor = conn.cursor() #默认获取的每行数据为元组，多个数据组成嵌套试元组 cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) #每行数据作为一个字典出现，多个数据字典组成列表 cursor.execute("select * from students") # rown = cursor.fetchmany(3)#只获取三条数据 rowall = cursor.fetchall() # print('只获取三条数据：',rown) print('获取全部数据',rowall) for i in rowall: print(i) conn.commit() #提交保存 cursor.close() #关闭光标 conn.close()#关闭连接
#!/usr/bin/env python # Author:zhangmingda import pymysql #向数据库中写入数据 conn = pymysql.connect(host='120.92.133.227',port=3306,user='zhangmingda',passwd='Wyf@1314',db='oldboy_db') cursor = conn.cursor() cursor.execute("delete from students where name='Zahangmiaochen' and 5< id <9;") #删除符合条件的数据from表中 cursor.execute("update students set age = 1 where id >= %s",(4,)) #更改原有数据 cursor.execute("insert into students (name,age,sex)values(%s,%s,%s)",('Zhangmiaochen',1,'girl'))#插入新数据 cursor.executemany("insert into students (name,age,sex)values(%s,%s,%s)",[('Zhangmiaochen',1,'girl'),('Gongzhu',1,'girl')])#插入多行新数据 conn.commit() #提交保存，否则数据库不会更改生效 cursor.close() conn.close() print('自增最新ID',cursor.lastrowid)