专栏首页有趣的djangoSQLAlchemy SQLAlchemy

SQLAlchemy SQLAlchemy

SQLAlchemy

1.1.SQLAlchemy介绍

SQLAlchemy是一个基于Python实现的ORM框架。该框架建立在 DB API之上,使用关系对象映射进行数据库操作,简言之便是:将类和对象转换成SQL,然后使用数据API执行SQL并获取执行结果。

安装

pip install sqlalchemy

组成部分

Engine,框架的引擎
Connection Pooling ,数据库连接池
Dialect,选择连接数据库的DB API种类
Schema/Types,架构和类型
SQL Exprression Language,SQL表达式语言

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

1.2.SQLAlchemy表结构

 (1)创建单表

import datetime
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String, Text, ForeignKey, DateTime, UniqueConstraint, Index

Base = declarative_base()


class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=False)
    email = Column(String(32), unique=True)
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

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


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间,否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收(重置)
    )

    Base.metadata.create_all(engine)


def drop_db():
    """
    根据类删除数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间,否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收(重置)
    )

    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_db()
    init_db()

(2)创建多个表(包含FK,M2M关系)

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

Base = declarative_base()


# ##################### 单表示例 #########################
class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    email = Column(String(32), unique=True)
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

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


class Hosts(Base):
    __tablename__ = 'hosts'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    ctime = Column(DateTime, default=datetime.datetime.now)


# ##################### 一对多示例 #########################
class Hobby(Base):
    '''爱好'''
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    caption = Column(String(50), default='篮球')


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.id"))    #hobby指的__tablename__ = 'hobby',而不是类名Hobby

    # 与生成表结构无关,仅用于查询方便
    hobby = relationship("Hobby", backref='pers')


# ##################### 多对多示例 #########################

#第三张表要自己生成
class Server2Group(Base):
    __tablename__ = 'server2group'
    id = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))


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

    # 与生成表结构无关,仅用于查询方便
    servers = relationship('Server', secondary='server2group', backref='groups')


class Server(Base):
    __tablename__ = 'server'

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


def init_db():
    """
    根据类创建数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间,否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收(重置)
    )

    Base.metadata.create_all(engine)


def drop_db():
    """
    根据类删除数据库表
    :return:
    """
    engine = create_engine(
        "mysql+pymysql://root:123@127.0.0.1:3306/s6?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间,否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收(重置)
    )

    Base.metadata.drop_all(engine)


if __name__ == '__main__':
    drop_db()
    init_db()

实例:

models.py

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

Base = declarative_base()


# ##################### 单表示例 #########################
class Users(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    age = Column(Integer, default=18)
    email = Column(String(32), unique=True)
    ctime = Column(DateTime, default=datetime.datetime.now)
    extra = Column(Text, nullable=True)

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


class Hosts(Base):
    __tablename__ = 'hosts'

    id = Column(Integer, primary_key=True)
    name = Column(String(32), index=True)
    ctime = Column(DateTime, default=datetime.datetime.now)


# ##################### 一对多示例 #########################
class Hobby(Base):
    '''爱好'''
    __tablename__ = 'hobby'
    id = Column(Integer, primary_key=True)
    caption = Column(String(50), default='篮球')


class Person(Base):
    __tablename__ = 'person'
    nid = Column(Integer, primary_key=True)
    name = Column(String(32), index=True, nullable=True)
    hobby_id = Column(Integer, ForeignKey("hobby.id"))    #hobby指的__tablename__ = 'hobby',而不是类名Hobby

    # 与生成表结构无关,仅用于查询方便
    hobby = relationship("Hobby", backref='pers')


# ##################### 多对多示例 #########################

#第三张表要自己生成
class Server2Group(Base):
    __tablename__ = 'server2group'
    id = Column(Integer, primary_key=True, autoincrement=True)
    server_id = Column(Integer, ForeignKey('server.id'))
    group_id = Column(Integer, ForeignKey('group.id'))


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

    # 与生成表结构无关,仅用于查询方便
    servers = relationship('Server', secondary='server2group', backref='groups')


class Server(Base):
    __tablename__ = 'server'

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

if __name__ == '__main__':
    engine = create_engine(
        "mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8",
        max_overflow=0,  # 超过连接池大小外最多创建的连接
        pool_size=5,  # 连接池大小
        pool_timeout=30,  # 池中没有线程最多等待的时间,否则报错
        pool_recycle=-1  # 多久之后对线程池中的线程进行一次连接的回收(重置)
    )

    Base.metadata.create_all(engine)   #创建

    # Base.metadata.drop_all(engine)   #删除

1.3.SQLAlchemy两种连接方式

第一种

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
import models

#1.创建连接池
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

#2.从连接池中获取数据库连接
session = Session()

#3.执行ORM操作
obj1 = models.Users(name="derek",email='derek@163.com')
session.add(obj1)
# 提交事务
session.commit()

# 4.关闭数据库连接(将连接放回连接池)
session.close()

第二种、基于scoped_session实现线程安全

首先导入,然后只要修改session = scoped_session(Session)

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
import models

#1.创建连接池
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

#2.从连接池中获取数据库连接
# session = Session()
session = scoped_session(Session)

#3.执行ORM操作
obj1 = models.Users(name="jack",email='jack@163.com')
session.add(obj1)
# 提交事务
session.commit()

# 4.关闭数据库连接
session.close()

1.4.增加数据

from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
import models

#1.创建连接池
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

#2.从连接池中获取数据库连接
session = Session()
# session = scoped_session(Session)

# #3.执行ORM操作
# obj1 = models.Users(name="jack",email='jack@163.com')
# session.add(obj1)
# # 提交事务
# session.commit()

#批量增加,里面是列表
session.add_all([
    models.Users(name="jack1",email='jack1@163.com'),
    models.Users(name="jack2",email='jack2@163.com'),
])
session.commit()

# 4.关闭数据库连接
session.close()

1.5.查看和删除数据

#查看数据
user_list = session.query(models.Users).all()
for row in user_list:
    print(row.id)
    print(row.name)
    print(row.email)
    print(row.ctime)
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
import models

#1.创建连接池
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

#2.从连接池中获取数据库连接
session = Session()
# session = scoped_session(Session)

# #3.执行ORM操作
# obj1 = models.Users(name="jack",email='jack@163.com')
# session.add(obj1)
# # 提交事务
# session.commit()

# #批量增加,里面是列表
# session.add_all([
#     models.Users(name="jack1",email='jack1@163.com'),
#     models.Users(name="jack2",email='jack2@163.com'),
# ])
# session.commit()

#查看数据
user_list = session.query(models.Users).all()
for row in user_list:
    print(row.id)
    print(row.name)
    print(row.email)
    print(row.ctime)

# 4.关闭数据库连接
session.close()

查看的结果:

添加过滤条件

user_list = session.query(models.Users).filter(models.Users.id > 2)   #id大于2的

 删除数据

#删除数据
session.query(models.Users).filter(models.Users.id > 4).delete()
session.commit()
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine
from sqlalchemy.orm import scoped_session
import models

#1.创建连接池
engine = create_engine("mysql+pymysql://root:123456@127.0.0.1:3306/test?charset=utf8", max_overflow=0, pool_size=5)
Session = sessionmaker(bind=engine)

#2.从连接池中获取数据库连接
session = Session()
# session = scoped_session(Session)

# #3.执行ORM操作
# obj1 = models.Users(name="jack",email='jack@163.com')
# session.add(obj1)
# # 提交事务
# session.commit()

# #批量增加,里面是列表
# session.add_all([
#     models.Users(name="jack1",email='jack1@163.com'),
#     models.Users(name="jack2",email='jack2@163.com'),
# ])
# session.commit()

#查看数据
# user_list = session.query(models.Users).all()
# user_list = session.query(models.Users).filter(models.Users.id > 2)   #id大于2的
# for row in user_list:
#     print(row.id)
#     print(row.name)
#     print(row.email)
#     print(row.ctime)

#删除数据
session.query(models.Users).filter(models.Users.id > 4).delete()
session.commit()


# 4.关闭数据库连接
session.close()

1.6.修改数据

#修改数据
session.query(models.Users).filter(models.Users.id == 4).update({'name':'Tom'})
session.query(models.Users).filter(models.Users.id == 1).update({'name': models.Users.name + "099"}, synchronize_session=False)
session.query(models.Users).filter(models.Users.id == 3).update({"age": models.Users.age + 1}, synchronize_session="evaluate") 
session.commit()

结果:

本文参与腾讯云自媒体分享计划,欢迎正在阅读的你也加入,一起分享。

我来说两句

0 条评论
登录 后参与评论

相关文章

  • BBS论坛(二十三)

    zhang_derek
  • Django+xadmin打造在线教育平台(十)

    代码 github下载 十四、xadmin的进阶开发 14.1.权限管理 (1)用户权限 超级用户拥有所有权限,其它添加的用户默认没有任何权限 进后台添加一个...

    zhang_derek
  • 微信小程序实战–集阅读与电影于一体的小程序项目(六)

    zhang_derek
  • Javascript操作将session资料存入window.name里

    查了一些资料,大家一致认为除了 Node.js 和服务端,在 JavaScript 里没有 session 这种东西(或者说很不常见),所有的变数,函式等等的资...

    德顺
  • 简化 SQL 递归查询

    自引用类型的表结构处理起来比较麻烦,比如“分类”表,通常包括自己的ID和父分类ID,当我们要做父分类路径、子分类路径之类的查询时很不方便,例如我们会使用嵌套查询...

    dys
  • 文件加密移动与解密(winform版

    如果你有一些文件不想让别人知道,并且不想将其从你的电脑删除,那么用这个简单的小程序就够了。。

    洋仔聊编程
  • 校招实习要抓紧

    前段时候有个公众号的读者问我,我们公司是否还招实习生。对此,我只能很抱歉的告诉他,我们的校招实习生已经完成了,现在的实习生已经在我们公司实习了好几个月了。

    飞雪无情
  • GraphQL在SAP Kyma中的广泛应用

    Jerry Wang
  • Sentence-BERT: 一种能快速计算句子相似度的孪生网络

    BERT和RoBERTa在文本语义相似度等句子对的回归任务上,已经达到了SOTA的结果。但是,它们都需要把两个句子同时喂到网络中,这样会导致巨大的计算开销:从1...

    zenRRan
  • java之mybatis之配置文件讲解

    Vincent-yuan

扫码关注云+社区

领取腾讯云代金券