本文主要Sqlalchemy的ORM components进行介绍
参考:SQLAlchemy 1.4 Documentation(中文版)

基本架构

SQLAlchemy由几个组件组成,包括SQL expression language和ORM。
为了实现这些组件,SQLAlchemy还提供了一个Engine类和MetaData类。

  • Engine–管理SQLAlchemy连接池和独立于数据库的SQL方言层。
  • MetaData – 用于收集和组织有关您的表布局(数据库模式)的信息。
  • SQL expression language–提供了一个API来对你的表执行查询和更新。
    所有这些都来自Python,而且都是以一种独立于数据库的方式。(本文不作介绍)
  • ORM – 提供了一种方便的方式来为你的Python对象添加数据库持久性,而不需要再添加其他的功能。

ORM创建表格

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


# 建立一个引擎对象,如果数据库不存在,将创建数据库,下面以sqlite为例
engine = create_engine("sqlite:///test.db", echo=True)
# 创造基类的实例,通常只会有一个实例,建议做单例模式
Base = declarative_base(bind=engine)


class Customers(Base):
    """在Base的基础上定义映射类

    Args:
        Base : 基类
    """    
    __tablename__ = 'customers'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    address = Column(String)
    email = Column(String)


def create_table():
    """创建还没有被创建的表
    """    
    Base.metadata.create_all() 

获取交互句柄

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

engine = create_engine("sqlite:///test.db", echo=True)

def get_session():
    """获取与数据库交互的句柄

    Returns:
        与数据库交互的句柄
    """    
    Session = sessionmaker(bind=engine)
    session = Session()
    return session

数据库操作

以新增customer为例

def add_customer(**kwargs):
    c = Customers(**kwargs)
    session = get_seesion()
    session.add(c)
    session.commit()
    session.close()
    return c

add_customer(name='xiaofan', address='xiaozhufeng', email='xiaofan@163.com')
其它常用操作
  • add_all()
  • query()
session.query(Customers).all()  # return list

session.query(Customers).first()  # return scalar
 # 超过一个报错“MultipleResultsFound”, 少于一个报错“NoResultFound”
session.query(Customers).filter(Customers.id == 1).one()   # retrun scalar

  • filter()
# return list
session.query(Customers).filter(Customers.name=='yasin')  # ==
session.query(Customers).filter(Customers.id>2)  # >
session.query(Customers).filter(Customers.name.like('Ra%'))  # like
session.query(Customers).filter(Customers.id.in_([1,3]))  # in
session.query(Customers).filter(Customers.id>2,
Customers.name.like('Ra%'))  # and
session.query(Customers).filter(and_(Customers.id>2,
Customers.name.like('Ra%')))  # and
session.query(Customers).filter(or_(Customers.id>2,
Customers.name.like('Ra%')))  # or
  • scalar()
session.query(Customers).filter(Customers.id==3).scalar()  # 仅当len(list)=1时有效
  • update()
session.query(Customers).filter(Customers.id!=2).update({Customers.name:"Mr."+C
ustomers.name}, synchronize_session=False)

Textual SQL

  • filter中使用SQL查询语句
from sqlalchemy import text

session.query(Customers).filter(text("id
  • 纯SQL语句查询
session.query(Customers).from_statement(text("SELECT * FROM customers")).all()

stmt = text("SELECT name, id, name, address, email FROM customers")
# name, id 对应链接到了Customers.id和Customers.name,互换了
stmt=stmt.columns(Customers.id, Customers.name) 
session.query(Customers.id, Customers.name).from_statement(stmt).all()

文章来源于互联网:Sqlalchemy ORM

发表评论