码迷，mamicode.com
首页 > 数据库 > 详细

pymysql和sqlalchemy模块操作mysql数据库

时间：2018-06-18 13:24:45      阅读：247      评论：0      收藏：0      [点我收藏+]

标签：ddr   sql   mysql事务   模块   prime   通道   records   col   数据库连接   

这两个模块可以直接通过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:123456@localhost/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:123456@localhost/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:123456@localhost/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:123456@localhost/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:123456@localhost/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:123456@localhost/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:123456@localhost/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:123456@localhost/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()

　　

 

pymysql和sqlalchemy模块操作mysql数据库

标签：ddr   sql   mysql事务   模块   prime   通道   records   col   数据库连接   

原文地址：https://www.cnblogs.com/ericbai/p/9194902.html
踩
(0)
赞
(0)
   
举报
评论 一句话评论（0）
登录后才能评论！
分享档案
更多>
2021年07月29日 (22)
2021年07月28日 (40)
2021年07月27日 (32)
2021年07月26日 (79)
2021年07月23日 (29)
2021年07月22日 (30)
2021年07月21日 (42)
2021年07月20日 (16)
2021年07月19日 (90)
2021年07月16日 (35)
周排行
mamicode.com排行更多图片
更多
友情链接
兰亭集智   国之画   百度统计   站长统计   阿里云   chrome插件   新版天听网
关于我们 - 联系我们 - 留言反馈
© 2014 mamicode.com 版权所有  联系我们:gaon5@hotmail.com
迷上了代码！