SQLAlchemy

Posted 2020-11-19 kakawith

一、sqlalchemy

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:pass@host:port/dbname[?key=value&key=value...]
 

使用 Schema Type/SQL Expression Language/Engine/ConnectionPooling/Dialect 进行数据库操作。Engine使用Schema Type创建一个特定的结构对象，之后通过SQL Expression Language将该对象转换成SQL语句，然后通过 ConnectionPooling 连接数据库，再然后通过 Dialect 执行SQL，并获取结果。

 

创建表

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:123456@127.0.0.1:3306/db1?charset=utf8", max_overflow=5)
Base = declarative_base()  # 生成orm基类


# 创建单表
class User(Base):
    __tablename__ = \'users\'  # 表名
    id = Column(Integer, primary_key=True, autoincrement=True)  # 主键自增
    name = Column(String(50), index=True, nullable=True)  # 创建索引并不能为空
    extra = Column(String(50))

    __table_args__ = (
        UniqueConstraint(\'id\', \'name\', name=\'uix_id_name\'),  # 联合唯一索引
        Index(\'ix_id_name\', \'name\', \'extra\'),  # 联合索引
    )


# 一对多
class Colour(Base):
    __tablename__ = \'colour\'  # 表名
    cid = Column(Integer, primary_key=True, autoincrement=True)
    colour = Column(String(20), server_default=\'red\', unique=True)


class Dress(Base):
    __tablename__ = \'dress\'  # 表名
    did = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    colour_id = Column(Integer, ForeignKey(Colour.cid))


# 多对多
class Group(Base):
    __tablename__ = \'group\'
    id = Column(Integer, primary_key=True)
    name = Column(String(64), unique=True, nullable=False)


class Server(Base):
    __tablename__ = \'server\'
    id = Column(Integer, primary_key=True, autoincrement=True)
    hostname = Column(String(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)  # 创建表结构


init_db()

Foreignkey的2种写法的区别:

使用第一种方法要注意顺序，第二种不用。

user_type_id = Column(Integer, ForeignKey(UserType.user_type_id))  # 使用这个写法，必须把UserType表放在前面
user_type_id = Column(Integer, ForeignKey(\'user_type.user_type_id\'))

也可以在类下面加上这个方法:

目的是在我们查询的时候，都显示的是方法，如果把该方法加入到相关类下面，返回的就是数据了

    def __repr__(self):
        tmp = \'%s - %s - %s - %s\' % (self.id, self.name, self.extra, self.num)
        return tmp

删除表

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

二、最基本的增删改查

# 1. 增加
# obj = Users(name=\'alex\')
# session.add(obj)
# session.commit()

# session.add_all([
#         Users(name=\'小egon\'),
#         Users(name=\'eric\')
# ])
# session.commit()

# 2. 查
# result = session.query(Users).all()
# for row in result:
#         print(row.id,row.name)

# result = session.query(Users).filter(Users.id >= 2)
# for row in result:
#         print(row.id,row.name)

# result = session.query(Users).filter(Users.id >= 2).first()
# print(result)

# 3.删
# session.query(Users).filter(Users.id >= 2).delete()
# session.commit()

# 4.改
# session.query(Users).filter(Users.id == 4).update({Users.name:\'egon\'})
# session.query(Users).filter(Users.id == 4).update({\'name\':\'小egon\'})
# session.query(Users).filter(Users.id == 4).update({\'name\':Users.name+"dsb"},synchronize_session=False)
# session.commit()