Użyj tabeli tymczasowej z SQLAlchemy

Question

Próbuję użyć użyć tabeli tymczasowej z SQLAlchemy i dołączyć do istniejącej tabeli. To, co mam tak daleko

engine = db.get_engine(db.app, 'MY_DATABASE') 
df = pd.DataFrame({"id": [1, 2, 3], "value": [100, 200, 300], "date": [date.today(), date.today(), date.today()]}) 
temp_table = db.Table('#temp_table', 
         db.Column('id', db.Integer), 
         db.Column('value', db.Integer), 
         db.Column('date', db.DateTime)) 
temp_table.create(engine) 
df.to_sql(name='tempdb.dbo.#temp_table', 
      con=engine, 
      if_exists='append', 
      index=False) 
query = db.session.query(ExistingTable.id).join(temp_table, temp_table.c.id == ExistingTable.id) 
out_df = pd.read_sql(query.statement, engine) 
temp_table.drop(engine) 
return out_df.to_dict('records') 

to nie zwraca żadnych wyników, ponieważ INSERT że to_sql robi nie dostać run (myślę, że to dlatego, że są prowadzone za pomocą sp_prepexec, ale nie jestem całkowicie na ten temat pewny).

Następnie próbowałem po prostu wypisać instrukcję SQL (CREATE TABLE #temp_table..., INSERT INTO #temp_table..., SELECT [id] FROM...), a następnie uruchomić pd.read_sql(query, engine). Otrzymuję komunikat o błędzie

Ten obiekt wyników nie zwraca wierszy. Został zamknięty automatycznie.

Zgaduję, że to dlatego, że oświadczenie ma więcej niż tylko SELECT?

W jaki sposób mogę rozwiązać ten problem (albo rozwiązanie zadziała, chociaż pierwsze byłoby lepsze, ponieważ unika zakodowanego kodu SQL). Aby było jasne, nie mogę zmodyfikować schematu w istniejącej bazie danych - jest to baza danych dostawcy.

Answer 1

W przypadku, gdy liczba rekordów do umieszczenia w tabeli tymczasowej jest małe/umiarkowane, jedną z możliwości byłoby użyć literal subquery lub values CTE zamiast tworzenia tabeli tymczasowej.

# MODEL 
class ExistingTable(Base): 
    __tablename__ = 'existing_table' 
    id = sa.Column(sa.Integer, primary_key=True) 
    name = sa.Column(sa.String) 
    # ... 

Załóżmy również następujące dane do wstawienia do temp tabeli:

# 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)), 
] 

Tworzenie CTE lub sub-zapytanie zawierające te dane:

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 

Tworzenie końcowe są za pośrednictwem wymagane połączenia i filtry:

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)  

Wreszcie ramka danych pandy:

out_df = pd.read_sql(query.statement, engine) 
result = out_df.to_dict('records') 

Answer 2

Można próbować używać innego rozwiązania - Proces nadwozia Tabela

Tabela proces-wpust jest po prostu stała tabela służy jako stół Temp. Aby umożliwić procesom jednoczesne korzystanie z tabeli, tabela ma dodatkową kolumnę do identyfikacji procesu. Najprostszym sposobem na jest globalna zmienna @@ spid (@@ spid to identyfikator procesu w serwerze SQL ).

...

Jedną z alternatyw dla procesów-kluczem jest użycie GUID (typ danych uniqueidentifier).

http://www.sommarskog.se/share_data.html#prockeyed