pymysql和sqlalchemy模块操作mysql数据库

Posted 2020-11-19 与君同悦

这两个模块可以直接通过pip安装：

打开mysql创建几个表：

CREATE TABLE `student_age`(
    `id` int not null,
    `age` int not null,
    `sex` ENUM(‘M‘, ‘F‘) not null,
    PRIMARY KEY(`id`)
);

INSERT INTO student_age(id, age, sex) VALUES(1, 18, ‘M‘), (2, 26, ‘M‘), (3, 20, ‘F‘);

CREATE TABLE `student_name`(
    `id` int not null auto_increment,
    `name` varchar(10) not null default ‘‘,
    `stu_id` int not null,
    PRIMARY KEY(`id`),
    FOREIGN KEY(`stu_id`) REFERENCES `student_age`(`id`)
);

INSERT INTO student_name(name, stu_id) VALUES(‘Jack‘, 1), (‘Eric‘, 2), (‘Alice‘, 3);

　　

pymysql基本连接示例：

import pymysql

#建立管道
conn = pymysql.connect(host=‘127.0.0.1‘, port=3306, user=‘root‘, passwd=‘123456‘, db=‘test‘)
#创建游标
cursor = conn.cursor()

effect_rows = cursor.execute(‘SELECT * FROM student_name‘)

print(‘rows[%d]‘ % effect_rows, cursor.fetchall())

#关闭连接
conn.close()

　　pymysql插入数据和事物的效果一样，可以实现回滚，自增id占用，必须提交才会生效：

import pymysql, time

#设置数据库连接参数
host = ‘localhost‘
port = 3306
user = ‘root‘
passwd = ‘123456‘
db = ‘test‘

#创建通道
conn = pymysql.connect(host=host, port=port, user=user, passwd=passwd, db=db)
#创建游标
cursor = conn.cursor()
#写sql语句
sql = "INSERT INTO student_age(id, age, sex) VALUES(4, 21, ‘F‘)"
#执行sql语句
effect_row = cursor.execute(sql)
#打印影响行
print(effect_row)
time.sleep(30)
#提交数据
conn.commit()
#关闭通道
conn.close()

　　orm操作数据库新建一张表：

‘‘‘
对象关系映射（英语：(Object Relational Mapping，简称ORM，或O/RM，或O/R mapping），是一种程序技术，用于实现面向对象编程语言里不同类型系统的数据之间的转换  。从效果上说，它其实是创建了一个可在编程语言里使用的--“虚拟对象数据库”。
‘‘‘
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding="utf-8", echo=True)

#生成orm基类
Base = declarative_base()

class User(Base):
    #表名
    __tablename__ = ‘user‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

#创建表结构
Base.metadata.create_all(engine)

　　新增数据：

import sqlalchemy
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://root:[email protected]/test", encoding=‘utf-8‘, echo=True)
#创建基本处理类
Base = declarative_base()
#
class User(Base):
    #表名
    __tablename__ = ‘user‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))
    
#绑定要操作的数据库
Session_class = sessionmaker(bind=engine)
#类似于创建游标
Session = Session_class()
obj1 = User(id=1, name=‘summer‘, password=‘111111‘)
#加入会话任务列表
session.add(obj1)
#此时数据还未真正写入
print(obj1.id, obj1.name, obj1.password)
#写入数据库
session.commit()

　　查询和格式化输出：

import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import sessionmaker
from sqlalchemy import func
#创建工程
engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding=‘utf8‘)
#创建基本处理类
Base = declarative_base()
#
class User(Base):
    #表名
    __tablename__ = ‘user‘
    id = Column(Integer, primary_key=True)
    name = Column(String(32))
    password = Column(String(64))
    
    #格式化输出查询结果
    def __repr__(self):
        return "<%s name:%s>" % (self.id, self.name)
    
#绑定要操作的数据库
Session_class = sessionmaker(bind=engine)
#类似于创建游标
Session = Session_class()
#查询
# data = Session.query(User).all()    #查询全部

#条件查询
data = Session.query(User).filter(User.id>1).first()   #获取第一个
#修改
data.name = ‘Tom‘
data = Session.query(User).filter(User.id>1).first()
#撤销操作
# Session.rollback()
# print(‘after rollback‘)
# data = Session.query(User).filter(User.id>1).first()

#提交之后才会作用到数据库，和mysql事务的的效果一样
# Session.commit()

# data = Session.query(User).filter(User.id>1).filter(User.id<4).all()  #获取区间数据

#统计
data = Session.query(User).filter(User.name.in_([‘Eric‘, ‘Alice‘])).count()

#分组查询
data = Session.query(func.count(User.name), User.name).group_by(User.name).all()
print(data)
#没有重写__repr__之前的访问方式
# print(data[0].id, data[0].name, data[0].password)

　　外键表的创建：

"""
外键表的创建：
学生表students
课程表days
签到表records，关联学生和课程表
"""
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, Enum

#创建工程
engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding=‘utf-8‘)
#创建基类
Base = declarative_base()

#创建表结构
class Student(Base):
    __tablename__ = ‘students‘
    id = Column(Integer, nullable=False, primary_key=True)
    name = Column(String(10), nullable=False)
    sex = Column(Enum(‘F‘, ‘M‘))
    
    def __repr__(self):
        return "<id:%s name:%s>" % (self.id, self.name)

class Days(Base):
    __tablename__ = ‘days‘
    id = Column(Integer, nullable=False, primary_key=True)
    content = Column(String(32), nullable=False)
    
    def __repr__(self):
        return "<day_id:%s content:%s>" % (self.id, self.content)
    
class Record(Base):
    __tablename__ = ‘records‘
    id = Column(Integer, nullable=False, primary_key=True)
    stu_id = Column(Integer, ForeignKey("students.id"), nullable=False)
    day_id = Column(Integer, ForeignKey("days.id"), nullable=False)
    status = Column(Enum("Yes", "No"))
    
    def __repr__(self):
        return "<id:%s status:%s>" % (self.id, self.status)

#生成表
Base.metadata.create_all(engine)

　　外键表的数据插入：

"""
外键表的创建：
学生表students
课程表days
签到表records，关联学生和课程表
"""
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, Enum

#创建工程
engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding=‘utf-8‘)
#创建基类
Base = declarative_base()

#创建表结构
class Student(Base):
    __tablename__ = ‘students‘
    id = Column(Integer, nullable=False, primary_key=True)
    name = Column(String(10), nullable=False)
    sex = Column(Enum(‘F‘, ‘M‘))
    
    def __repr__(self):
        return "<id:%s name:%s>" % (self.id, self.name)

class Days(Base):
    __tablename__ = ‘days‘
    id = Column(Integer, nullable=False, primary_key=True)
    content = Column(String(32), nullable=False)
    
    def __repr__(self):
        return "<day_id:%s content:%s>" % (self.id, self.content)
    
class Record(Base):
    __tablename__ = ‘records‘
    id = Column(Integer, nullable=False, primary_key=True)
    stu_id = Column(Integer, ForeignKey("students.id"), nullable=False)
    day_id = Column(Integer, ForeignKey("days.id"), nullable=False)
    status = Column(Enum("Yes", "No"))
    
    def __repr__(self):
        return "<id:%s status:%s>" % (self.id, self.status)

#生成表
Base.metadata.create_all(engine)

　　外键表的关联查询：

"""
外键关联查询
"""
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import Column, Integer, String, Enum, ForeignKey

#创建工程
engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding=‘utf-8‘)
#创建基类
Base = declarative_base()
#创建表结构
class Students(Base):
    __tablename__ = ‘students‘
    id = Column(Integer, nullable=False, primary_key=True)
    name = Column(String(10), nullable=False)
    sex = Column(Enum(‘F‘, ‘M‘))
    
    def __repr__(self):
        return "<id:%s name:%s>" % (self.id, self.name)
    
class Days(Base):
    __tablename__ = ‘days‘
    id = Column(Integer, nullable=False, primary_key=True)
    content = Column(String(32), nullable=False)
    
    def __repr__(self):
        return "<id:%s content:%s>" % (self.id, self.content)

class Records(Base):
    __tablename__ = ‘records‘
    id = Column(Integer, nullable=False, primary_key=True)
    stu_id = Column(Integer, ForeignKey("students.id"), nullable=False)
    day_id = Column(Integer, ForeignKey("days.id"), nullable=False)
    status = Column(Enum("Yes", "No"))
    students = relationship("Students", backref="students_records")
    days = relationship("Days", backref="days_records")
    def __repr__(self):
        return "<name:%s content:%s status:%s>" % (self.students.name, self.days.content, self.status)

#生成表
Base.metadata.create_all(engine)
#创建会话
Session_class = sessionmaker(bind=engine)
#创建游标
session = Session_class()
data = session.query(Students).filter(Students.name==‘Eric‘).first()
print(data.students_records)

　　两个外键关联到同一张表：

"""
两个外键关联到同一张表
"""
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer, ForeignKey
from sqlalchemy.orm import relationship

engine = create_engine("mysql+pymysql://root:[email protected]/test", encoding=‘utf-8‘)
Base = declarative_base()

class Address(Base):
    __tablename__ = ‘address‘
    id = Column(Integer, primary_key=True, nullable=False)
    province = Column(String(10), nullable=True)
    city = Column(String(10), nullable=True)
    
class Users(Base):
    __tablename__ = ‘users‘
    id = Column(Integer, primary_key=True, nullable=False)
    name = Column(String(10), nullable=True)
    first_addr_id = Column(Integer, ForeignKey(‘address.id‘))
    second_addr_id = Column(Integer, ForeignKey(‘address.id‘))
    first_addr = relationship("Address", foreign_keys=[first_addr_id])
    second_addr = relationship("Address", foreign_keys=[second_addr_id])
    
    def __repr__(self):
        return "<name %s address %s %s>" % (self.name, self.first_addr.province, self.second_addr.province)
    
Base.metadata.create_all(engine)

　　写入数据：

"""
往表中写入数据
"""
import ex5_1
from sqlalchemy.orm import sessionmaker

Session_class = sessionmaker(bind=ex5_1.engine)
session = Session_class()

add1 = ex5_1.Address(province=‘HeNan‘, city=‘NanYang‘)
add2 = ex5_1.Address(province=‘HeBei‘, city=‘HanDan‘)
add3 = ex5_1.Address(province=‘HuNan‘, city=‘YueYang‘)

session.add_all([add1, add2, add3])

user1 = ex5_1.Users(name=‘Eric‘, first_addr=add1, second_addr=add2)
user2 = ex5_1.Users(name=‘Alice‘, first_addr=add2, second_addr=add3)
user3 = ex5_1.Users(name=‘Peter‘, first_addr=add3, second_addr=add1)

session.add_all([user1, user2, user3])
session.commit()

　　查询：

import ex5_1

data = ex5_1.session.query(ex5_1.Users).filter(ex5_1.Users.name==‘Eric‘).first()
print(data)

　　多对多外键：

#创建表
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, String, Integer, DATE, Table, ForeignKey
from sqlalchemy.orm import relationship

engine = create_engine("mysql+pymysql://root:[email protected]/test", 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 Books(Base):
    __tablename__ = ‘books‘
    id = Column(Integer, primary_key=True, nullable=True)
    book_name = Column(String(20), nullable=False)
    publish_time = Column(DATE)
    #创建到authors表的映射，关联关系表book_m2m_author，回查字段books
    authors = relationship("Authors", secondary=book_m2m_author, backref="books")
    
    def __repr__(self):
        return "<book_name:%s>" % self.book_name
    
class Authors(Base):
    __tablename__ = ‘authors‘
    id = Column(Integer, primary_key=True, nullable=True)
    author_name = Column(String(20), nullable=False)
    def __repr__(self):
        return "<author_name:%s>" % self.author_name

Base.metadata.create_all(engine)

　　创建数据：

#插入数据
import ex6_1
from sqlalchemy.orm import sessionmaker

#创建会话
Session_class = sessionmaker(bind=ex6_1.engine)
Session = Session_class()

b1 = ex6_1.Books(book_name=‘C++ primer plus‘)
b2 = ex6_1.Books(book_name=‘Python‘)
b3 = ex6_1.Books(book_name=‘Java‘)
a1 = ex6_1.Authors(author_name=‘Eric‘)
a2 = ex6_1.Authors(author_name=‘Alice‘)
a3 = ex6_1.Authors(author_name=‘James‘)

b1.authors = [a1, a2]
b2.authors = [a2, a3]
b3.authors = [a3, a1]

Session.add_all([b1, b2, b3, a1, a2, a3])
Session.commit()

　　关联查询和删除操作：

#查询数据
import ex6_1
from sqlalchemy.orm import sessionmaker

Session_class = sessionmaker(bind=ex6_1.engine)
Session = Session_class()

data = Session.query(ex6_1.Books).filter(ex6_1.Books.book_name==‘Python‘).first()
print("book_name:%s author:%s" % (data, data.authors))

#删除数据时不用管book_m2m_author，sqlalchemy会自动删除

#通过书删除作者
# book_obj = Session.query(ex6_1.Books).filter(ex6_1.Books.book_name==‘Python‘).first()
# author_obj = Session.query(ex6_1.Authors).filter(ex6_1.Authors.author_name==‘James‘).first()
# book_obj.authors.remove(author_obj)

#删除作者时，会把这个作者跟所有书的关联关系数据也自动删除
data = Session.query(ex6_1.Authors).filter(ex6_1.Authors.author_name==‘Eric‘).first()
#从authors里面删除这个作者，并从book_m2m_author里面所有书中删除这个作者
Session.delete(data)

#提交修改到数据库
Session.commit()