SqlAlchemy ORM

Posted 2020-10-24 ︻◣_蝸犇り~

ORM介绍

orm英文全称object relational mapping,就是对象映射关系程序，简单来说我们类似python这种面向对象的程序来说一切皆对象，但是我们使用的数据库却都是关系型的，为了保证一致的使用习惯，通过orm将编程语言的对象模型和数据库的关系模型建立映射关系，这样我们在使用编程语言对数据库进行操作的时候可以直接使用编程语言的对象模型进行操作就可以了，而不用直接使用sql语言。

orm的优点：

1. 隐藏了数据访问细节，“封闭”的通用数据库交互，ORM的核心。他使得我们的通用数据库交互变得简单易行，并且完全不用考虑该死的SQL语句。快速开发，由此而来。
2. ORM使我们构造固化数据结构变得简单易行。

缺点：

1. 无可避免的，自动化意味着映射和关联管理，代价是牺牲性能（早期，这是所有不喜欢ORM人的共同点）。现在的各种ORM框架都在尝试使用各种方法来减轻这块（LazyLoad，Cache），效果还是很显著的。

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果

Dialect用于和数据API进行交流，根据配置文件的不同调用不同的数据库API，从而实现对数据库的操作

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:password@host:port/dbname[?key=value&key=value...]
	
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

ORM创建表结构：

方法1：

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column,Integer,String

# create_engine() 初始化数据库连接
# \'数据库类型+数据库驱动名称://用户名:口令@机器地址:端口号/数据库名\'
# charset=utf-8 设置表支持中文字符、echo=True打印创建过程
engine = create_engine("mysql+pymysql://root:123456@localhost/oldboy?charset=utf-8",
                       encoding="utf-8",echo=True)

Base = declarative_base()   # 创建orm的基类

# 定义User对象
class User(Base):
    __tablename__ = "user"   # 表的名字

    # 表的结构
    id = Column(Integer,primary_key=True)
    name = Column(String(32))
    password = Column(String(64))

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

方法二：

from sqlalchemy import Table,MetaData,Column,Integer,String,create_engine
from sqlalchemy.orm import mapper

engine = create_engine("mysql+pymysql://root:123456@localhost/oldboy?charset=utf8",echo=True) 

metadata = MetaData()

user = Table("user",metadata,
             Column("id",Integer,primary_key=True),
             Column("name",String(50)),
             Column("fullname",String(50)),
             Column("password",String(12))
             )

class User(object):
    def __init__(self,name,fullname,password):
        self.name = name
        self.fullname = fullname
        self.password = password

mapper(User,user)  # 创建映射关系

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

插入数据：

Session_class = sessionmaker(bind=engine)  # 创建与数据库的会话session class,返回的是个class,不是实例
Session = Session_class()   # 生成session实例

user_obj = User(name="alex",password="alex3714")  # 生成要创建的数据对象
print(user_obj.name,user_obj.id)  # 此时还没创建对象,user_obj.id返回None

Session.add(user_obj)  # 把要创建的数据对象添加到session里,统一创建
print(user_obj.name,user_obj.id)  # 此时也依然没创建

Session.commit()  # 统一提交,创建数据

查询数据：

# query：查询对象、filter_by：过滤条件、first：显示第一个

my_user = Session.query(User).filter_by(name="alex").first()

print(my_user)   # 返回的是内存地址<__main__.User object at 0x7f92d89ef518>

在User类中加入__repr__函数，可将内存地址中相应数据返回

# 重构__repr__方法后，不管直接输出对象还是通过print打印的信息都按我们__repr__方法中定义的格式进行显示

def __repr__(self):   # __repr__ 和 __str__这两个方法都是用于显示的,__str__是面向用户的，而__repr__面向程序员
　　return "<%s(name=%s,password=%s)>" %(self.__class__.name,self.name,self.password)

修改数据：

my_user = Session.query(User).filter_by(name="alex").filter_by(password="sb1234").first()  # 多个条件时可加多个filter_by

my_user.name="Alex Li"    # 修改mysql_user中name的值

Session.commit()  # 提交

事务回滚：

my_user = Session.query(User).filter_by(id=1).first()  # 查id=1的数据
my_user.name = "Jack"  # 修改id=1数据中的name

fake_user = User(name="Rain",password="123456")  # 插入数据
Session.add(fake_user)  # 要插入的数据添加到Session里面

# filter与filter_by都是过滤条件,用法类似,里面的条件语句不同
print(Session.query(User).filter(User.name.in_(["Jack","rain"])).all()) # 查询出所有User对象中name为Jack、rain的

Session.rollback()  # 事务回滚

print(Session.query(User).filter(User.name.in_(["Jack","rain"])).all())

简单语句：

# all()获取所有数据
Session.query(User.name,User.id).all()

# 多条件查询
Session.query(User).filter(User.id>1).filter(User.id<7).all()

# 统计次数
Session.query(User).filter(User.name.like("Ra%")).count()

# 分组
from sqlalchemy import func

Session.query(User.name,func.count(User.name)).group_by(User.name).all() # 按User.name分组

相当于sql: select user.name,count(user.name) from user group by user.name;

外键关联：

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

engine = create_engine("mysql+pymysql://root:123456@localhost/oldboy?charset=utf8",
                       encoding="utf8")

Base = declarative_base()  # 创建ORM基类

session_class = sessionmaker(bind=engine)  # 对表的操作交给session_class来操作,返回的是一个class
session = session_class()  # 创建一个session对象,在session中完成多个事务


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

    def __repr__(self):
        return "<User(name=%s,password=%s)>" %(self.name,self.password)

class Address(Base):
    __tablename__ = "addresses"
    id = Column(Integer,primary_key=True)
    email_address = Column(String(32),nullable=False)
	#注意点：增加用户时user_id只能是表user中存在的id,否则会报错
    user_id = Column(Integer,ForeignKey("user.id"))  # addresses表中的外键关联user表中的id
	
	# relationship函数是sqlalchemy对关系之间提供的一种便利的调用方式
    user = relationship("User",backref="addresses")  # 通过user调用user表中的变量, user表中通过addresses调用addresses表中的变量

    def __repr__(self):
        return "<Address(email_address=%s)>" %self.email_address


Base.metadata.create_all(engine)   # 构造表结构

obj = session.query(User).all()[0]  # 查询user表语句

print("obj.addresses:",obj.addresses)

obj.addresses = [Address(email_address="[e-mail]"),
                 Address(email_address="[e-mail]")]  # 通过user对象反查关联的addresses插入数据

for i in obj.addresses:  # 通过user对象反查关联的addresses记录
　　print(i)

session.commit()  # 事务提交

addr_obj = session.query(Address).all()[-1]  # 查看address表中的数据
print(addr_obj.user)  # 通过user查看user表中的数据

常用 查询语法：

	equals:
		query.filter(User.name == "ed")
	not equals:
		query.filter(User.name != "ed")
	like:
		query.filter(User.name.like("%ed%"))
	in:
		query.filter(User.name.in_(["ed","wendy","jack"]))
	not in:
		query.filter(~User.name.in_(["ed","wendy","jack"]))
	is null / is not null:
		query.filter(User.name == None)
		query.filter(User.name.is_(None))
		query.filter(User.name != None)
		query.filter(User.name.isnot(None))
	and:
		from sqlalchemy import and_
		query.filter(and_(User.name == "ed",User.fullname == "Ed Jones"))
		query.filter(User.name == "ed",User.fullname == "Ed Jones")
		query.filter(User.name == "ed").filter(User.fullname == "Ed Jones")
	or:
		from sqlalchemy import or_
		query.filter(or_(User.name == "ed",User.fullname == "Ed Jones"))
	match:
		query.filter(User.name.match("wendy"))

多外键关联：（Customer表有2个字段都关联了Address表）

 1 from sqlalchemy import Integer,ForeignKey,String,Column
 2 from sqlalchemy.ext.declarative import declarative_base
 3 from sqlalchemy.orm import relationship
 4 from sqlalchemy import create_engine
 5 
 6 engine = create_engine("mysql+pymysql://root:123456@localhost/oldboy?charset=utf8")
 7 
 8 Base = declarative_base()
 9 
10 
11 class Customer(Base):
12     __tablename__ = "customer"
13     id = Column(Integer,primary_key=True)
14     name = Column(String(32))
15 
16     billing_address_id = Column(Integer,ForeignKey("address.id"))
17     shopping_address_id = Column(Integer,ForeignKey("address.id"))
18 
19     bill_address = relationship("Address",foregin_keys=[billing_address_id])
20     shopping_address = relationship("Address",foregin_keys=[shopping_address_id])
21 
22 
23 class Address(Base):
24     __tablename__ = "address"
25     id = Column(Integer,primary_key=True)
26     street = Column(String(32))
27     city = Column(String(32))
28     state = Column(String(32))
29 
30 
31 Base.metadata.create_all(engine)

多外键关联

多对多关系：

 1 from sqlalchemy import Table,Column,Integer,String,DATE,ForeignKey
 2 from sqlalchemy.orm import relationship
 3 from sqlalchemy.ext.declarative import declarative_base
 4 from sqlalchemy import create_engine
 5 from sqlalchemy.orm import sessionmaker
 6 
 7 engine = create_engine("mysql+pymysql://root:123456@localhost/oldboy?charset=utf8",
 8                        encoding="utf8")
 9 
10 Base = declarative_base()
11 
12 # 外键关联表,ORM自动维护,不需要手动维护的一般采用Table这种方式创建
13 book_m2m_author = Table("book_m2m_author",Base.metadata,
14                         Column("book_id",Integer,ForeignKey("books.id")),
15                         Column("author_id",Integer,ForeignKey("authors.id"))
16                         )
17 
18 class Book(Base):
19     __tablename__ = "books"
20     id = Column(Integer,primary_key=True)
21     name = Column(String(32))
22     pub_date = Column(DATE)
23     authors = relationship("Author",secondary=book_m2m_author,backref="books")
24 
25     def __repr__(self):
26         return self.name
27 
28 
29 class Author(Base):
30     __tablename__ = "authors"
31     id = Column(Integer,primary_key=True)
32     name = Column(String(32))
33 
34     def __repr__(self):
35         return self.name
36 
37 Base.metadata.create_all(engine)
38 
39 
40 session_class = sessionmaker(bind=engine)
41 s = session_class()
42 
43 b1 = Book(name="Python自动化")
44 b2 = Book(name="人工智能")
45 b3 = Book(name="心灵鸡汤")
46 b4 = Book(name="读者")
47 
48 a1 = Author(name="Alex")
49 a2 = Author(name="Jack")
50 a3 = Author(name="Rain")
51 
52 b1.authors = [a1,a2]
53 b2.authors = [a1,a2,a3]
54 
55 s.add_all([b1,b2,b3,a1,a2,a3])
56 
57 s.commit()
58 
59 # ------------ 书名查作者 ----------------
60 obj = s.query(Book).filter(Book.name.like("%ython%")).first()
61 print(obj.name,obj.authors)
62 
63 # ------------ 作者查书名 ----------------
64 author_obj = s.query(Author).filter_by(name="Alex").first()
65 print(author_obj.name,author_obj.books)

ORM多对多

多对多删除

删除数据时不用管boo_m2m_authors　， sqlalchemy会自动帮你把对应的数据删除

通过书删除作者

删除数据时不用管boo_m2m_authors　， sqlalchemy会自动帮你把对应的数据删除

通过书删除作者

author_obj =s.query(Author).filter_by(name="Jack").first()
 
book_obj = s.query(Book).filter_by(name="跟Alex学把妹").first()
 
book_obj.authors.remove(author_obj) #从一本书里删除一个作者

s.commit()

直接删除作者　

删除作者时，会把这个作者跟所有书的关联关系数据也自动删除

author_obj =s.query(Author).filter_by(name="Alex").first()

# print(author_obj.name , author_obj.books)

s.delete(author_obj)

s.commit()　

 

处理中文

sqlalchemy设置编码字符集一定要在数据库访问的URL上增加charset=utf8，否则数据库的连接就不是utf8的编码格式

eng = create_engine(\'mysql://root:root@localhost:3306/test2?charset=utf8\',echo=True)

 

补充知识点：

• #  排序 
• users = session.query(User).order_by(User.name) 
• #  降序（需要导入desc方法） 
• from sqlalchemy import desc 
• users = session.query(User).order_by(desc(User.name))