sqlalchemy的基础使用，sqlalchemy调用外部创建的触发器

http://blog.csdn.net/u013600225/article/details/53913550

这个链接举了一个触发器的例子，这个例子在SQLiteStudio上面是可以正常运行的（在Navicat下执行失败），然后就想试验一下python的sqlalchemy能不能正常执行外部创建的这些触发器，然后就写了下面这个例子：

# -*- coding: utf-8 -*-
# 前面的是几个通用函数.
import sqlalchemy
from sqlalchemy.ext.declarative import declarative_base
BaseMeta = declarative_base()  # BaseMeta.metadata.tables.keys() 含有"收集到的"所有的表


def initialize_session(s):
    """initialize_session和terminate_session要配对使用"""
    if isinstance(s, sqlalchemy.orm.session.sessionmaker):
        session = s()
        needClose = True
    elif isinstance(s, sqlalchemy.orm.session.Session):
        session = s
        needClose = False
    else:
        raise ValueError("未知的s={}".format(type(s)))
    return session, needClose


def terminate_session(session, needClose):
    """initialize_session和terminate_session要配对使用"""
    if needClose:
        session.close()
    return None


def _upsert(s, allNode):
    """allNode需要是一个可迭代的类型,可以是list,tuple等"""
    session, needClose = initialize_session(s)
    try:
        for node in allNode:
            session.merge(node)
        session.commit()
    finally:
        terminate_session(session, needClose)
    return len(allNode)


class data_part1(BaseMeta):
    __tablename__ = "data_part1"
    exchange = sqlalchemy.Column(sqlalchemy.String(16), primary_key=True)
    code = sqlalchemy.Column(sqlalchemy.String(32), primary_key=True)
    name = sqlalchemy.Column(sqlalchemy.String(32))
    group_id = sqlalchemy.Column(sqlalchemy.String(32))


class data_part2(BaseMeta):
    __tablename__ = "data_part2"
    exchange = sqlalchemy.Column(sqlalchemy.String(16), primary_key=True)
    code = sqlalchemy.Column(sqlalchemy.String(32), primary_key=True)
    price_tick = sqlalchemy.Column(sqlalchemy.Float)


class data_all(BaseMeta):
    __tablename__ = "data_all"
    exchange = sqlalchemy.Column(sqlalchemy.String(16), primary_key=True)
    code = sqlalchemy.Column(sqlalchemy.String(32), primary_key=True)
    name = sqlalchemy.Column(sqlalchemy.String(32))
    group_id = sqlalchemy.Column(sqlalchemy.String(32))
    price_tick = sqlalchemy.Column(sqlalchemy.Float)


if __name__ == "__main__":
    # Python 3.5.2 |Anaconda 4.2.0 (64-bit)
    # sqlalchemy.__version__ is '1.0.13'
    # 先创建好各个表和各个触发器,然后用sqlalchemy往表里面插入数据,结果显示,触发器是可以正常运行的.
    print(sqlalchemy.__version__)
    sqlite3_file = r"sqlite:///./_test.sqlite3"
    engine = sqlalchemy.create_engine(sqlite3_file)
    #
    BaseMeta.metadata.create_all(engine)
    #
    SessionMaker = sqlalchemy.orm.sessionmaker(bind=engine)
    session = SessionMaker()
    #
    if True:
        dp1 = data_part1()
        dp1.exchange = "SSE"
        dp1.code = "600000"
        dp1.name = 'PFYH'
        dp1.group_id = 'ASHARE'
        _upsert(session, [dp1])
        dp1.name = 'PUFA'
        _upsert(session, [dp1])
    if True:
        dp2 = data_part2()
        dp2.exchange = 'SSE'
        dp2.code = '600000'
        dp2.price_tick = 0.01
        _upsert(session, [dp2])
    #
    session.close()
    print("DONE")
    exit(0)

完。

更多相关文章

随机推荐