Python编程：orm之sqlalchemy模块

orm英文全称object relational mapping,对象映射关系
http://www.sqlalchemy.org/

常用操作

"""
MySQL-Python
    mysql+mysqldb://<user>:<password>@<host>[:<port>]/<dbname>

pymysql
    mysql+pymysql://<username>:<password>@<host>/<dbname>[?<options>]

MySQL-Connector
    mysql+mysqlconnector://<user>:<password>@<host>[:<port>]/<dbname>

cx_Oracle
    oracle+cx_oracle://user:[email protected]:port/dbname[?key=value&key=value...]

SQLite
    driver://user:[email protected]/database
"""


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", echo=False)

Base = declarative_base()  #声明基类

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

    def __repr__(self):
        return "<%s name: %s >" %(self.id, self.name)

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

# 创建一条数据
Session_class = sessionmaker(bind=engine)  # 创建会话类
session = Session_class()  # 实例化

# user_obj = User(name="Tom", password="123456") # 生成数据对象
# print(user_obj.name, user_obj.id) # id=None

# 添加数据
# session.add(user_obj)
# print(user_obj.name, user_obj.id)
#
# session.commit()  # 提交事务

# 查询
data = session.query(User).filter(User.id==2).first()  # 或者all()
print(data)

# 修改
data.name = "Alex"
data.password = "abc"
session.commit()

# 多条件查询
data = session.query(User).filter(User.id>2).filter(User.name =="alex").all()  # 或者all()
print(data)

# 回滚
user1= User(name="xiaobai", password="xxx")
session.add(user1)
data1 = session.query(User).filter(User.name == "xiaobai").all()
print(data1)

session.rollback()
data2 = session.query(User).filter(User.name == "xiaobai").all()
print(data2)

# 统计
count = session.query(User).filter(User.name.like("a%")).count()
print(count)

# 分组
data = session.query(User.name, func.count(User.name)).group_by(User.name).all()
print(data)

外键关联

import sqlalchemy  # 第三方库，需要安装
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DATE, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import func

# 创建表结构
engine = create_engine("mysql+pymysql://root:[email protected]/test",
                       encoding="utf-8", echo=False)

Base = declarative_base()  #声明基类

class Student(Base):
    __tablename__ = "student"
    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 study_record(Base):
    __tablename__ = "study_record"
    id = Column(Integer, primary_key=True)
    day = Column(Integer, nullable=False)
    status = Column(String(32), nullable=False)
    stu_id = Column(Integer, ForeignKey("student.id"))

    # 允许在study_record表中，通过backref字段反向查出student的关联项
    student = relationship("student", backref="study_record")

Base.metadata.create_all(engine)

一对多关系

# 如果一个User拥有多个Book，就可以定义一对多关系
class User(Base):
    __tablename__ = 'user'

    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # 一对多:
    books = relationship('Book')

class Book(Base):
    __tablename__ = 'book'

    id = Column(String(20), primary_key=True)
    name = Column(String(20))
    # “多”的一方的book表是通过外键关联到user表的:
    user_id = Column(String(20), ForeignKey('user.id'))

创建多外键表结构

# orm_mfk_api.py

# 为使用者提供统一的数据结构接口

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, DATE, ForeignKey
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()

class User(Base):
    __tablename__ = "user"
    id = Column(Integer, primary_key=True)
    name = Column(String(32), nullable=False)
    bill_address_id = Column(Integer, ForeignKey("address.id"))
    ship_address_id = Column(Integer, ForeignKey("address.id"))

    # 一对多:
    bill_address = relationship("Address", foreign_keys=[bill_address_id])
    ship_address = relationship("Address", foreign_keys=[ship_address_id])

    def __repr__(self):
        return "id: %s, name: %s, bill: %s, ship: %s" %(
            self.id, self.name, self.bill_address_id, self.ship_address_id)

class Address(Base):
    __tablename__ = "address"
    id = Column(Integer, primary_key=True)
    city = Column(String(32), nullable=False)

# 设置引擎，创建表
engine = create_engine("mysql+pymysql://root:[email protected]/test",
                       encoding="utf-8", echo=False)
Base.metadata.create_all(engine)

# 调用多外键关联的api，对数据进行增删改查

from orm_mfk_api import engine, User, Address
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(engine)
session = Session()

# 插入地址
addr1 = Address(city="beijing")
addr2 = Address(city="wuhang")
addr3 = Address(city="lanzhou")
addr4 = Address(city="dali")
# session.add_all([addr1, addr2, addr3, addr4])

# 插入用户
user1 = User(name="Tom", bill_address=addr1, ship_address=addr2)
user2 = User(name="Jack", bill_address=addr1, ship_address=addr1)
user3 = User(name="Jimi", bill_address=addr3, ship_address=addr2)
# session.add_all([user1, user2, user3])

session.commit()

# 查询
result =session.query(User).filter(User.name=="Jimi").first()
print(result)

多对多关系

# orm_m2m_api.py

# 图书与作者
#多对多关系的统一接口
from sqlalchemy import Integer, String, Column, Table, DATE, ForeignKey
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import  declarative_base
from sqlalchemy.orm import relationship

# 处理中文字符
engine = create_engine("mysql+pymysql://root:[email protected]/test?charset=utf8")
Base = declarative_base()

# 创建关系表，第三张表连接book和author
book2author = Table("book2author", Base.metadata,
                    Column("book_id", Integer, ForeignKey("books.id")),
                    Column("author_id", Integer, ForeignKey("authors.id"))
                    )

class Author(Base):
    __tablename__ = "authors"
    id = Column(Integer,primary_key=True, autoincrement=True)
    name = Column(String(32))

    def __repr__(self):
        return self.name

class Book(Base):
    __tablename__ = "books"
    id = Column(Integer, primary_key=True, autoincrement=True)
    name = Column(String(64))
    pub_date = Column(DATE)

    authors = relationship("Author", secondary=book2author, backref="books")

    def __repr__(self):
        return self.name

Base.metadata.create_all(engine)

# 调用多对多接口，管理作者与图书的数据

import orm_m2m_api
from sqlalchemy.orm import sessionmaker

Session = sessionmaker(orm_m2m_api.engine)
session = Session()

# 作者
a1 = orm_m2m_api.Author(name="Tom")
a2 = orm_m2m_api.Author(name="Jack")
a3 = orm_m2m_api.Author(name="Jimi")
a4 = orm_m2m_api.Author(name="Ben")

# 图书
b1 = orm_m2m_api.Book(name="lear python", pub_date="2018-12-13")
b2 = orm_m2m_api.Book(name="lear java", pub_date="2018-12-14")
b3 = orm_m2m_api.Book(name="lear cpp", pub_date="2018-12-15")
b4 = orm_m2m_api.Book(name="中文书籍", pub_date="2018-12-15")

# 设置图书与作者的关系
b1.authors=[a1, a2]
b2.authors=[a1, a3]
b3.authors=[a4]

# 提交数据
# session.add_all([a1, a2, a3, a4, b1, b2, b3])
# session.commit()

# 书查询作者
result = session.query(orm_m2m_api.Book).filter(orm_m2m_api.Book.id==2).first()
print(result, result.authors)  # lear java [Tom, Jimi]


# 作者查询书
result = session.query(orm_m2m_api.Author).filter(orm_m2m_api.Author.id==1).first()
print(result, result.books)  # Tom [lear python, lear java]

# 删除书的作者
a5 = session.query(orm_m2m_api.Author).filter(orm_m2m_api.Author.id==1).first()
b5 = session.query(orm_m2m_api.Book).filter(orm_m2m_api.Book.id==1).first()
# b5.authors.remove(a5)
# session.commit()

# 删除作者
a6 = session.query(orm_m2m_api.Author).filter(orm_m2m_api.Author.id==4).first()
session.delete(a6)
session.commit()

完整示例代码：
《学生管理系统》
https://github.com/mouday/StudentManagerSys