ORM框架SQLAlchemy

介绍

SQLAlchemy是Python编程语言下的一款ORM框架，该框架建立在数据库API之上，使用关系对象映射进行数据库操作，

简言之便是：将对象转换成SQL，然后使用数据API执行SQL并获取执行结果。

1，安装：

 pip3 install SQLAlchemy 

SQLAlchemy本身无法操作数据库，其必须以来pymsql等第三方插件，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:pass@host:port/dbname[?key=value&key=value...]
   
更多详见：http://docs.sqlalchemy.org/en/latest/dialects/index.html

 ORM功能使用

使用 ORM/Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 所有组件对数据进行操作。根据类创建对象，对象转换成SQL，执行SQL。

 

1、创建表

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, ForeignKey, UniqueConstraint, Index
from sqlalchemy.orm import sessionmaker, relationship
from sqlalchemy import create_engine
 
#连接mysql
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)
 
Base = declarative_base()
 
# 创建单表 
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(VARCHAR(32))
    extra = Column(VARCHAR(16))
 
    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )
 
 
# 一对多 
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(VARCHAR(50), default='red', unique=True)
 
 
class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(VARCHAR(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
 
  
# 多对多 
class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(VARCHAR(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
 
 
class Server(Base):
    __tablename__ = 'server'
 
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(VARCHAR(64), unique=True, nullable=False)
 
 
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
  
def init_db():
    Base.metadata.create_all(engine)

def drop_db():
    Base.metadata.drop_all(engine)

 

2，操作表：

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

engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/t1", max_overflow=5)

Base = declarative_base()

# 创建单表 
class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(VARCHAR(32))
    extra = Column(VARCHAR(16))

    __table_args__ = (
    UniqueConstraint('id', 'name', name='uix_id_name'),
        Index('ix_id_name', 'name', 'extra'),
    )

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

# 一对多 
class Favor(Base):
    __tablename__ = 'favor'
    nid = Column(Integer, primary_key=True)
    caption = Column(VARCHAR(50), default='red', unique=True)

    def __repr__(self):
        return "%s-%s" %(self.nid, self.caption)

class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    favor_id = Column(Integer, ForeignKey("favor.nid"))
    # 与生成表结构无关，仅用于查询方便
    favor = relationship("Favor", backref='pers')

# 多对多 
class ServerToGroup(Base):
    __tablename__ = 'servertogroup'
    nid = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))
    group = relationship("Group", backref='s2g')
    server = relationship("Server", backref='s2g')

class Group(Base):
    __tablename__ = 'group'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)
    port = Column(Integer, default=22)
    # group = relationship('Group',secondary=ServerToGroup,backref='host_list')


class Server(Base):
    __tablename__ = 'server'

    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(64), unique=True, nullable=False)


def init_db():
    Base.metadata.create_all(engine)

def drop_db():
    Base.metadata.drop_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

3，案例： 

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

### 连接MySQL 
engine = create_engine("mysql+pymysql://root:123@127.0.0.1:3306/db3?charset=utf8", max_overflow=5)

Base = declarative_base()


'''
create table  users(
    id int auto_increment primary key,
    name varchar(32) not null default '',
    extra varchar(32) not null default ''
)
'''

'''
Usertype
id    title      xxoo
1     普通用户
row.xxoo : 多条记录对象
'''
class UserType(Base):
    __tablename__ = 'usertype'  ### 表名
    id = Column(Integer, autoincrement=True, primary_key=True)
    title = Column(String(32), nullable=False, server_default='')

'''
users
id     name   extra  type_id
1      zekai   nb      1
usertype = releationship('Usertype')
row.usertype
'''

class Users(Base):
    __tablename__ = 'users'
    id = Column(Integer, autoincrement=True, primary_key=True)
    name = Column(String(32), nullable=False, server_default='')
    extra = Column(String(32), nullable=False, server_default='')
    type_id = Column(Integer, ForeignKey(UserType.id))
    usertype = relationship('UserType', backref='xxoo')

    #创建索引
    __table_args__ = (
        UniqueConstraint('id', 'name', name='uix_id_name'),  ### 联合唯一索引
        Index('ix_name_extra', 'name', 'extra'),  ### 组合索引
    )


# 删除表 
def drop_db():
    Base.metadata.drop_all(engine)

# 创建表 
def create_db():
    ## 会将当前执行文件中所有继承自Base类的类,生成表
    Base.metadata.create_all(engine)

drop_db()
create_db()


###操作表中的数据
Session = sessionmaker(bind=engine)
session = Session()


### 增加一条数据
obj = UserType(title='普通用户')
session.add(obj)

### 添加多条数据
session.add_all([
    UserType(title='VIP用户'),
    UserType(title='VIP中P用户'),
    UserType(title='SVIP用户'),
    UserType(title='黑金用户')
])

### 查询 
# ### 查询全部  返回的是一个列表, 列表中套对象
res = session.query(UserType).all()
for row in res:
    print(row.id, row.name)

#### 查询一条数据 
res = session.query(UserType).first()
print(res)

# where条件 
res = session.query(UserType).filter(UserType.name=='VIP用户', UserType.id==2).all()
for row in res:
    print(row.id, row.name)
print(res[0].name, res[0].id)
res = session.query(UserType).filter_by(name='VIP用户').all()
print(res)

## 删除:
session.query(UserType).filter(UserType.id>3).delete()


## 修改
session.query(UserType).filter(UserType.id == 3).update({"name" : "SVIP用户"})


### MySQL 高级查询操作

#### 通配符  分组 分页 排序 between and in not in

### between ... and ...
res = session.query(UserType).filter(UserType.id.between(1,3)).all()
for row in res:
    print(row.id, row.title)


## in 操作  bool_ 
res = session.query(UserType).filter(UserType.id.in_([1,3,4])).all()
print(res)


###  not in
# select * from Usertype
res = session.query(UserType).filter(~UserType.id.in_([1,3,4])).all()
for row in res:
    print(row.id, row.title)

session.query(UserType).filter(Users.id.in_(session.query(Users.id).filter_by(name='eric'))).all()
res = session.query(UserType.title).filter(~UserType.id.in_([1,3,4])).all()
for row in res:
    print( row.title)

from sqlalchemy import and_, or_


### 通配符
ret = session.query(UserType).all()

from sqlalchemy.sql import func
res = session.query(
    Users.type_id,
    func.max(Users.id),
    func.min(Users.id)).group_by(Users.type_id).all()

print(res)
ret = session.query(
    func.max(Users.id),
    func.min(Users.id)).group_by(Users.type_id).having(func.min(Users.id) >2).all()

### left join


#############1. 查询某一个用户的用户类型##############

### 第一种方法:
res = session.query(Users,UserType).join(UserType, isouter=True).all()
# print(res)
for row in res:
    print(row[0].id, row[0].name, row[1].title)

### 第二种方法
res = session.query(Users).all()
for row in res:
    print(row.id, row.name, row.extra, row.usertype.title)


##############2. 某一个类型下面的用户###############

### 第一种方法
res = session.query(UserType).all()
for row in res:
    print(row.id, row.title, session.query(Users).filter(Users.type_id == row.id).all() )

### 第二种方法(正向|反向查询)
res = session.query(UserType).all()
for row in res:
    print(row.id, row.title, row.xxoo)



session.commit()
session.close()