文章目录
MySQL基础
略……
参考https://www.runoob.com/mysql/mysql-tutorial.html
pymysql
pymsql是Python中个操作MySQL的模块，它的使用方法和另一个MySQL操作模块MySQLdb几乎相同。
pymysql基本操作
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob import pymysql # 创建mysql连接 conn = pymysql.connect(host="192.168.2.114", port=3306, user="root", password="12345678", db="python_test") cursor = conn.cursor() # Create a new cursor to execute queries with. # 创建数据表 cursor.execute(''' CREATE TABLE IF NOT EXISTS `person` ( `id` int(0) NOT NULL AUTO_INCREMENT, `name` varchar(255) NULL, `age` smallint(8) NULL, PRIMARY KEY (`id`)); ''') # 查询 rows = cursor.execute("SELECT * FROM `person`;") # 执行查询操作，返回匹配行数 print("Total rows: ", rows) # 插入 cursor.execute("INSERT INTO `person` (`name`, `age`) VALUES ('Tom', 22);") cursor.execute("INSERT INTO `person` (`name`, `age`) VALUES ('Jack', 23);") # 更新 update_rows = cursor.execute("UPDATE `person` SET `age`=30 where `id`<%s;", 2) # 传递参数到SQL语句 print("Updated rows: ", update_rows) # 执行批量插入，传入一个由字段元组组成的列表 cursor.executemany("INSERT INTO `person` (`name`, `age`) VALUES (%s, %s);", [("Frank", 38), ("Mack", 27)]) # 提交、断开 conn.commit() # 提交操作（默认开启了事务，所有操作需要提交才能正式写入） cursor.close() conn.close() # 关闭连接 |
pymysql获取查询数据
默认通过execute执行查询操作返回的是结果行数，获取结果需要使用cursor的fetchXXX方法。
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob import pymysql conn = pymysql.connect(host="192.168.2.114", port=3306, user="root", password="12345678", db="python_test") cursor = conn.cursor() # 默认获取的数据是元祖类型，如果想要字典类型的数据需要在创建cursor时声明 # cursor = conn.cursor(cursor=pymysql.cursors.DictCursor) cursor.execute("SELECT * FROM `person`;") print(cursor.fetchone()) # 获取第1行数据 print(cursor.fetchone()) # 再获取1行数据 print(cursor.fetchmany(2)) # 接着获取2行数据 print(cursor.fetchall()) # 获取剩下的所有数据 cursor.close() conn.close() |
SQLAchemy
SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。
SQLAlchemy创建数据表、插入数据
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建连接引擎 # 创建基类 BaseClass = declarative_base() # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) # 整型可以省略长度使用默认值 # 创建表 BaseClass.metadata.create_all(engine) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() data1 = Student(name="Tommy", sex="M", age=28) # 实例化要插入的数据 data2 = Student(name="Jack", sex="M", age=27) session.add(data1) # 准备插入 session.add(data2) session.commit() # 提交，正式插入 |
SQLAlchemy查询
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) def __repr__(self): return str({"std_id": self.std_id, "name": self.name, "sex": self.sex, "age": self.age}) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() select_result_all = session.query(Student).filter_by().all() # 相当于select student.* from student，返回所有符合条件的类实例 print(select_result_all) # select_result_all对象是Student的实例，通过重构对象的__repr__方法来获取自定义的信息 select_filtered1 = session.query(Student.name, Student.age).filter(Student.age > 20).all() # 使用filter方法过滤 print(select_filtered1) select_filtered2 = session.query(Student.name, Student.age).filter_by(std_id=2).all() # 使用filter_by方法过滤 print(select_filtered2) # 多条件查询 select_filtered3 = session.query(Student.name, Student.age).filter(Student.age > 20).filter(Student.std_id > 1).all() print(select_filtered3) |
SQLAlchemy分组和统计
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger, func from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) def __repr__(self): return str({"std_id": self.std_id, "name": self.name, "sex": self.sex, "age": self.age}) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() # 相当于select student.sex, count(student.sex) from student group by student.sex result = session.query(Student.sex, func.count(Student.sex)).group_by(Student.sex).all() print(result) |
SQLAlchemy关联查询
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) def __repr__(self): return str({"std_id": self.std_id, "name": self.name, "sex": self.sex, "age": self.age}) class Score(BaseClass): __tablename__ = "score" id = Column(Integer, primary_key=True) uid = Column(Integer, ForeignKey("student.std_id")) # 通过表的列名建立外键关联 score = Column(SmallInteger, nullable=False, default=0) student = relationship(Student, backref="score") # 建立关系，允许在student表中相关的外键查询对应的record记录 def __repr__(self): return str({"id": self.id, "uid": self.uid, "score": self.score}) # 创建表 BaseClass.metadata.create_all(engine) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() result = session.query(Student.name, Score.score).filter(Student.std_id == Score.uid).all() # 手工联合字段查询 print(result) join_result = session.query(Student.name, Score.score).join(Score).all() # 自动根据外键关联联合查询 print(join_result) |
SQLAlchemy修改数据
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) def __repr__(self): return str({"std_id": self.std_id, "name": self.name, "sex": self.sex, "age": self.age}) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() select_student = session.query(Student).filter_by(std_id=2).first() # 查询并返回第一条结果 print(select_student) select_student.age = 20 # 修改对象属性 session.commit() # 提交 new_student = Student(name="Lincoln", age=26) # 创建新对象 session.add(new_student) # 加入session print(session.query(Student).filter(Student.name == "Lincoln").all()) # 查询结果 session.rollback() # 回滚 print(session.query(Student).filter(Student.name == "Lincoln").all()) # 再次查询 |
SQLAlchemy外键关联
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, SmallInteger, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 创建表对象 class Student(BaseClass): __tablename__ = "student" # 表名 std_id = Column(Integer, primary_key=True, ) # 主键 name = Column(String(50)) # 列 sex = Column(String(32)) age = Column(SmallInteger) def __repr__(self): return str({"std_id": self.std_id, "name": self.name, "sex": self.sex, "age": self.age}) class Course(BaseClass): __tablename__ = "course" cid = Column(Integer, primary_key=True) name = Column(String(50)) def __repr__(self): return str({"cid": self.cid, "name": self.name}) class Record(BaseClass): __tablename__ = "record" id = Column(Integer, primary_key=True) cid = Column(Integer, ForeignKey(Course.cid)) # 通过对象属性建立外键关联 uid = Column(Integer, ForeignKey("student.std_id")) # 通过表的列名建立外键关联 score = Column(SmallInteger, nullable=False, default=0) # 建立关系，可通过外键关联查询到关系表中的相关记录，同时允许在对应的表中通过backref指定的字段查询本表中的记录 student = relationship(Student, backref="record") course = relationship(Course, backref="record") def __repr__(self): return str({"student": self.student.name, "course": self.course.name, "score": self.score}) # 创建表 BaseClass.metadata.create_all(engine) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() result = session.query(Student).filter(Student.std_id == 3).first() for i in result.record: print(i) # print(session.query(Student).all()) # print(session.query(Course).all()) # print(session.query(Record).all()) |
SQLAlchemy多外键关联
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
#!/usr/bin/env python36 # -*- coding: utf-8 -*- # Author: Bob from sqlalchemy import create_engine, Column, Integer, String, Table, ForeignKey from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, relationship # 创建连接引擎 BaseClass = declarative_base() # 创建基类 # 书和作者映射表(直接创建对象而不是类，因为此表不需要手工插入数据，由ORM自动维护) BookAuthors = Table( 'book_authors', # 表明 BaseClass.metadata, # metadata Column('book_id', Integer, ForeignKey("book.id")), Column('author_id', Integer, ForeignKey("author.id")), ) # 创建表对象 class Author(BaseClass): __tablename__ = "author" id = Column(Integer, primary_key=True) name = Column(String(50)) email = Column(String(50)) def __repr__(self): return str({"id": self.id, "name": self.name, "email": self.email}) class Book(BaseClass): __tablename__ = "book" # 表名 id = Column(Integer, primary_key=True, ) # 主键 name = Column(String(50)) # 列 # 通过relationship authors = relationship(Author, secondary=BookAuthors, backref="books") def __repr__(self): return str({"id": self.id, "name": self.name}) # 创建表 BaseClass.metadata.drop_all(engine) BaseClass.metadata.create_all(engine) # 创建会话（连接）类并实例化 SessionClass = sessionmaker(bind=engine) session = SessionClass() b1 = Book(id=1, name="Python从入门到放弃") b2 = Book(id=2, name="Java从入门到放弃") b3 = Book(id=3, name="C++从入门到放弃") a1 = Author(id=1, name="张全蛋") a2 = Author(id=2, name="李二狗") a3 = Author(id=3, name="王尼玛") b1.authors = [a1, a2] b2.authors = [a1, a3] b3.authors = [a2, a3] session.add_all([a1, a2, a3, b1, b2, b3]) session.commit() # 查询所有表中的记录 # print(session.query(Author).all()) # print(session.query(Book).all()) # print(session.query(BookAuthors).all()) author = session.query(Author).filter(Author.id == 1).first() print("作者[%s], 参与书籍[%s]" % (author.name, author.books)) # 查询作者参与的书 book = session.query(Book).filter(Book.id == 1).first() print("书名[%s], 作者[%s]" % (book.name, book.authors)) # 查询书的作者 # 从指定书中删除作者 book.authors.remove(author) session.commit() print("从[%s]书中删除作者[%s]" % (book.name, author.name)) book = session.query(Book).filter(Book.id == 1).first() print("书名[%s], 作者[%s]" % (book.name, book.authors)) # 删除指定作者 book3 = session.query(Book).filter(Book.id == 3).first() print("书名[%s], 作者[%s]" % (book3.name, book3.authors)) author2 = session.query(Author).filter(Author.id == 2).first() session.delete(author2) session.commit() print("从数据库删除作者[%s]" % author2.name) book3 = session.query(Book).filter(Book.id == 3).first() print("书名[%s], 作者[%s]" % (book3.name, book3.authors)) |
原文链接：Python 从入门到放弃 - Lesson 12 mysql和ORM，转载请注明来源！