1.多张表映射到一个类
import pandas as pd
from settings import DATABASES
from sqlalchemy import create_engine
import sqlalchemy.orm
import sqlalchemy.ext.declarative
metadata = sqlalchemy.MetaData()
STK_BASIC_INFO_table = sqlalchemy.Table(‘quant_stk_basic_info‘, metadata,
sqlalchemy.Column(‘STK_UNI_CODE‘, sqlalchemy.Integer, primary_key = True),
sqlalchemy.Column(‘STK_CODE‘, sqlalchemy.String(23)),
)
FCT_VALUE_ROW_Q_table = sqlalchemy.Table(‘quant_fct_value_row_q‘,metadata,
sqlalchemy.Column(‘SEC_UNI_CODE‘, sqlalchemy.Integer,
sqlalchemy.ForeignKey(‘quant_stk_basic_info.STK_UNI_CODE‘),primary_key=True),
sqlalchemy.Column("END_DATE", sqlalchemy.Date),
sqlalchemy.Column("A7041000005", sqlalchemy.Float),
）
j_Q = sqlalchemy.join(STK_BASIC_INFO_table, FCT_VALUE_ROW_Q_table)
#首先需要生成一个BaseModel类，作为所有模型类的基类
BaseModel = sqlalchemy.ext.declarative.declarative_base()
#构建 数据模型 Factor
class Factor_Q(BaseModel):
__table__ = j_Q
gp_unicode = sqlalchemy.orm.column_property(STK_BASIC_INFO_table.c.STK_UNI_CODE, FCT_VALUE_ROW_Q_table.c.SEC_UNI_CODE)
gpcode = STK_BASIC_INFO_table.c.STK_CODE
date = FCT_VALUE_ROW_Q_table.c.END_DATE
ACC_PAY_TDAYS = FCT_VALUE_ROW_Q_table.c.A7041000005
# 利用Session对象链接数据库
self._engine_mysql = create_engine(
DATABASES["Quant"].get("engine") + "://" + DATABASES["Quant"].get("user") + ":" + DATABASES[
"Quant"].get("password") + "@" + DATABASES["Quant"].get("host") + "/" + DATABASES["Quant"].get(
"db") + "?charset=utf8")
DBSession = sqlalchemy.orm.sessionmaker(bind=self._engine_mysql) # 创建回话类
session = DBSession() # 创建回话对象
q = session.query(Factor_Q.ACC_PAY_TDAYS).filter(Factor_Q.gpcode.in_([‘002252.XSHE‘,‘002253.XSHE‘]))
for data in q:
print data
2.单张表映射到一个类
# 首先需要生成一个BaseModel类，作为所有模型类的基类
BaseModel = sqlalchemy.ext.declarative.declarative_base()
# 构建数据模型 FIN_IDX_ANA
class FIN_IDX_ANA(BaseModel):
__tablename__ = ‘fin_idx_ana‘ # 表名
__table_args__ = (
PrimaryKeyConstraint(‘COM_UNI_CODE‘, ‘END_DATE‘),
)
# 表结构
com_unicode = sqlalchemy.Column("COM_UNI_CODE", sqlalchemy.Integer)
gpcode = sqlalchemy.Column("STK_CODE", sqlalchemy.String(10), nullable=False)
date = sqlalchemy.Column("END_DATE", sqlalchemy.Date)
BEPS = sqlalchemy.Column("BEPS", sqlalchemy.Float, default=0)
之后的调用方法和上面1的一样。
Demo:
百度云（13207134391）
Python/sqlalchemy