pymysql-sqlalchemy-orm

场景：一个作者映射多个书籍，一个书籍有多个作者

作者表：

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()

#!/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)

#!/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)