sqlalchemy示例代码

多对多外键

场景：一个作者映射多个书籍，一个书籍有多个作者作者表：id name email   sex1  Alex alex@..  M2  Rail rail@..  W书籍表：id name1  book12  book2作者书籍映射关系表：id  book_id  anthor_id1   1           12   1           23   2           1【Alex同时参与了两本书的创作】【book1 有两个作者参与创作】

#!/usr/bin/env python# Author:zhangmingda''''''from sqlalchemy import create_engine,ForeignKey,DATE,Tablefrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import relationship,sessionmakerengine = 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.nameclass 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.nameBase.metadata.create_all(engine)

#!/usr/bin/env python# Author:zhangmingdaimport orm_many_to_many_foreign_key as m2mfrom sqlalchemy.orm import sessionmakerSessionClass = 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,DATEfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import relationship,sessionmakerengine = 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:zhangmingdafrom day12.多对一外键到同一张表 import foreginkey_many_ormfrom sqlalchemy.orm import sessionmakerSession_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 funcprint('分组查询统计：',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,DATEfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import relationship,sessionmakerengine = 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:zhangmingdaimport sqlalchemy'''表的创建 查询'''from sqlalchemy import create_engine,ForeignKeyfrom sqlalchemy.ext.declarative import declarative_basefrom sqlalchemy import Column, Integer, Stringfrom sqlalchemy.orm import relationship,sessionmakerengine = 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_addressBase.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:zhangmingdaimport 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:zhangmingdaimport 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)