[python] Flask-SQLAlchemy 앱에서 원시 SQL을 실행하는 방법

SQL Alchemy 세션 개체에는 고유 한 execute방법이 있습니다.

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

모든 응용 프로그램 쿼리는 원시 SQL인지 여부에 관계없이 세션 개체를 통과해야합니다. 이를 통해 트랜잭션 이 쿼리를 올바르게 관리 할 수 있으므로 동일한 요청의 여러 쿼리를 단일 단위로 커밋하거나 롤백 할 수 있습니다. 엔진 이나 연결 을 사용하여 트랜잭션 외부로 나가면 미묘한 위험에 노출되어 데이터가 손상 될 수있는 버그를 감지하기 어려울 수 있습니다. 각 요청은 하나의 트랜잭션에만 연결되어야하며이를 사용 db.session하면 애플리케이션에 해당하는 것입니다.

또한 매개 변수화 된 쿼리를execute 위해 설계되었습니다 . SQL 인젝션 공격으로부터 자신을 보호하기 위해 쿼리에 입력 할 때 예제 와 같이 매개 변수를 사용하십시오 . a 를 두 번째 인수로 전달하여 이러한 매개 변수의 값을 제공 할 수 있습니다 . 여기서 각 키는 조회에 표시되는 매개 변수의 이름입니다. 매개 변수 자체의 정확한 구문은 데이터베이스에 따라 다를 수 있지만 모든 주요 관계형 데이터베이스는이를 특정 형식으로 지원합니다.:valdict

그것의 가정 SELECT쿼리,이 반환됩니다 반복 가능 의 RowProxy객체.

다양한 기술로 개별 열에 액세스 할 수 있습니다.

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

개인적으로 결과를 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)

Flask-SQLAlchemy 확장을 사용하지 않는 경우에도 세션을 쉽게 사용할 수 있습니다.

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})

답변

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')

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>"))

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

export FLASK_APP=manage

flask shell

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

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