<p>如果要插入到临时表中的记录数很小/中等,一种可能是使用<code>literal subquery</code>或<code>values CTE</code>而不是创建临时表。</p>
<pre><code># MODEL
class ExistingTable(Base):
__tablename__ = 'existing_table'
id = sa.Column(sa.Integer, primary_key=True)
name = sa.Column(sa.String)
# ...
</code></pre>
<p>假设还要将以下数据插入到<code>temp</code>表中:</p>
<pre><code># This data retrieved from another database and used for filtering
rows = [
(1, 100, datetime.date(2017, 1, 1)),
(3, 300, datetime.date(2017, 3, 1)),
(5, 500, datetime.date(2017, 5, 1)),
]
</code></pre>
<p>创建包含该数据的CTE或子查询:</p>
<pre><code>stmts = [
# @NOTE: optimization to reduce the size of the statement:
# make type cast only for first row, for other rows DB engine will infer
sa.select([
sa.cast(sa.literal(i), sa.Integer).label("id"),
sa.cast(sa.literal(v), sa.Integer).label("value"),
sa.cast(sa.literal(d), sa.DateTime).label("date"),
]) if idx == 0 else
sa.select([sa.literal(i), sa.literal(v), sa.literal(d)]) # no type cast
for idx, (i, v, d) in enumerate(rows)
]
subquery = sa.union_all(*stmts)
# Choose one option below.
# I personally prefer B because one could reuse the CTE multiple times in the same query
# subquery = subquery.alias("temp_table") # option A
subquery = subquery.cte(name="temp_table") # option B
</code></pre>
<p>使用所需的联接和筛选器创建最终查询:</p>
<pre><code>query = (
session
.query(ExistingTable.id)
.join(subquery, subquery.c.id == ExistingTable.id)
# .filter(subquery.c.date >= XXX_DATE)
)
# TEMP: Test result output
for res in query:
print(res)
</code></pre>
<p>最后,获取pandas数据帧:</p>
<pre><code>out_df = pd.read_sql(query.statement, engine)
result = out_df.to_dict('records')
</code></pre>