Jak wykonać surowy SQL w aplikacji Flask-SQLAlchemy

Jak wykonać surowy SQL w SQLAlchemy?

Mam aplikacji sieci web Python, która działa na kolbie i interfejsów do bazy danych za pośrednictwem SQLAlchemy.

Potrzebuję sposobu na uruchomienie surowego SQL. Kwerenda obejmuje wiele połączeń tabel wraz z widokami Inline.

Próbowałem:

connection = db.session.connection()
connection.execute( <sql here> )

Ale ciągle pojawiają się błędy bramy.

Czy próbowałeś:

result = db.engine.execute("<sql here>")

lub:

from sqlalchemy import text

sql = text('select name from penguins')
result = db.engine.execute(sql)
names = [row[0] for row in result]
print names

Obiekty sesji Alchemia SQL mają własną executemetodę:

result = db.session.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

Wszystkie zapytania aplikacji powinny przechodzić przez obiekt sesji, bez względu na to, czy jest to surowy SQL, czy nie. Zapewnia to prawidłowe zarządzanie zapytaniami przez transakcję , co pozwala na zatwierdzenie lub wycofanie wielu zapytań w tym samym żądaniu jako pojedynczej jednostki. Wyjście poza transakcję za pomocą silnika lub połączenia naraża Cię na znacznie większe ryzyko subtelnych, być może trudnych do wykrycia błędów, które mogą spowodować uszkodzenie danych. Każde żądanie powinno być powiązane tylko z jedną transakcją, a użycie db.sessionzapewni, że tak będzie w przypadku Twojej aplikacji.

Weź również pod uwagę, że executejest przeznaczony do sparametryzowanych zapytań . Użyj parametrów, takich jak :valw przykładzie, do wszelkich danych wejściowych do zapytania, aby uchronić się przed atakami iniekcyjnymi SQL. Możesz podać wartość tych parametrów, przekazując dictjako drugi argument, gdzie każdy klucz jest nazwą parametru wyświetlaną w zapytaniu. Dokładna składnia samego parametru może być różna w zależności od bazy danych, ale wszystkie główne relacyjne bazy danych obsługują je w jakiejś formie.

Zakładając, że jest to SELECTzapytanie, zwróci iterowalną liczbę RowProxyobiektów.

Możesz uzyskać dostęp do poszczególnych kolumn za pomocą różnych technik:

for r in result:
    print(r[0]) # Access by positional index
    print(r['my_column']) # Access by column name as a string
    r_dict = dict(r.items()) # convert to dict keyed by column names

Osobiście wolę przekonwertować wyniki na namedtuples:

from collections import namedtuple

Record = namedtuple('Record', result.keys())
records = [Record(*r) for r in result.fetchall()]
for r in records:
    print(r.my_column)
    print(r)

Jeśli nie używasz rozszerzenia Flask-SQLAlchemy, nadal możesz łatwo użyć sesji:

import sqlalchemy
from sqlalchemy.orm import sessionmaker, scoped_session

engine = sqlalchemy.create_engine('my connection string')
Session = scoped_session(sessionmaker(bind=engine))

s = Session()
result = s.execute('SELECT * FROM my_table WHERE my_column = :val', {'val': 5})

docs: Samouczek języka SQL Expression - Korzystanie z tekstu

przykład:

from sqlalchemy.sql import text

connection = engine.connect()

# recommended
cmd = 'select * from Employees where EmployeeGroup = :group'
employeeGroup = 'Staff'
employees = connection.execute(text(cmd), group = employeeGroup)

# or - wee more difficult to interpret the command
employeeGroup = 'Staff'
employees = connection.execute(
                  text('select * from Employees where EmployeeGroup = :group'), 
                  group = employeeGroup)

# or - notice the requirement to quote 'Staff'
employees = connection.execute(
                  text("select * from Employees where EmployeeGroup = 'Staff'"))


for employee in employees: logger.debug(employee)
# output
(0, 'Tim', 'Gurra', 'Staff', '991-509-9284')
(1, 'Jim', 'Carey', 'Staff', '832-252-1910')
(2, 'Lee', 'Asher', 'Staff', '897-747-1564')
(3, 'Ben', 'Hayes', 'Staff', '584-255-2631')

Możesz uzyskać wyniki zapytań SELECT SQL za pomocą from_statement()i text()jak pokazano tutaj . W ten sposób nie musisz radzić sobie z krotkami. Jako przykład dla klasy Usero nazwie tabeli, usersktórą możesz wypróbować,

from sqlalchemy.sql import text
.
.
.
user = session.query(User).from_statement(
    text("SELECT * FROM users where name=:name")).\
    params(name='ed').all()

return user

result = db.engine.execute(text("<sql here>"))

wykonuje, <sql here>ale go nie zatwierdza, chyba że jesteś w autocommittrybie. Tak więc wstawki i aktualizacje nie odzwierciedlają się w bazie danych.

Aby zatwierdzić po zmianach, wykonaj

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

Jest to uproszczona odpowiedź na pytanie, jak uruchomić zapytanie SQL z Flask Shell

Najpierw zamapuj moduł (jeśli Twój moduł / aplikacja jest manage.py w folderze głównym, a jesteś w systemie operacyjnym UNIX), uruchom:

export FLASK_APP=manage

Uruchom pocisk Flask

flask shell

Zaimportuj to, czego potrzebujemy:

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
db = SQLAlchemy(app)
from sqlalchemy import text

Uruchom zapytanie:

result = db.engine.execute(text("<sql here>").execution_options(autocommit=True))

Wykorzystuje to obecnie połączenie z bazą danych, które ma aplikację.

Czy próbowałeś używać connection.execute(text( <sql here> ), <bind params here> )i wiązać parametry zgodnie z opisem w dokumentacji ? Pomoże to rozwiązać wiele problemów z formatowaniem parametrów i wydajnością. Być może błąd bramy przekroczył limit czasu? Parametry wiązania powodują, że złożone zapytania są wykonywane znacznie szybciej.

Jeśli chcesz uniknąć krotki, innym sposobem jest wywołanie first, oneani allmetod:

query = db.engine.execute("SELECT * FROM blogs "
                           "WHERE id = 1 ")

assert query.first().name == "Welcome to my blog"