sqlAlchemy를 사용한 저장 프로시저
sqlAlchemy를 사용하여 sql 서버의 저장 프로시저를 어떻게 호출할 수 있습니까?
엔진 및 연결부에는execute()
임의의 sql 문에 사용할 수 있는 메서드이며 세션에도 사용할 수 있습니다.예:
results = sess.execute('myproc ?, ?', [param1, param2])
사용할 수 있습니다.outparam()
필요한 경우 출력 매개 변수를 만듭니다(또는 바인딩 매개 변수 사용).bindparam()
와 함께isoutparam=True
옵션)
컨텍스트:저는 MySQL과 ORM 매핑 없이 플라스크-sqalchemy를 사용합니다.일반적으로 다음을 사용합니다.
# in the init method
_db = SqlAlchemy(app)
#... somewhere in my code ...
_db.session.execute(query)
저장 프로시저 호출은 즉시 지원되지 않습니다.callproc
일반적이지는 않지만 mysql 커넥터에 고유합니다.
매개 변수가 없는 저장 프로시저의 경우 다음과 같은 쿼리를 실행할 수 있습니다.
_db.session.execute(sqlalchemy.text("CALL my_proc(:param)"), param='something')
평소처럼.구급대원이 있을 때는 상황이 더 복잡해집니다
매개 변수를 사용하는 한 가지 방법은 기본 커넥터에 액세스하는 것입니다.engine.raw_connection()
예:
conn = _db.engine.raw_connection()
# do the call. The actual parameter does not matter, could be ['lala'] as well
results = conn.cursor().callproc('my_proc_with_one_out_param', [0])
conn.close() # commit
print(results) # will print (<out param result>)
이것은 우리가 out 매개변수에 접근할 수 있기 때문에 좋지만, 이 연결은 플라스크 세션에 의해 관리되지 않습니다.즉, 관리되는 다른 쿼리와 마찬가지로 커밋/중지되지 않습니다...(절차에 부작용이 있는 경우에만 문제가 있음).
결국 저는 이렇게 했습니다.
# do the call and store the result in a local mysql variabl
# the name does not matter, as long as it is prefixed by @
_db.session.execute('CALL my_proc_with_one_out_param(@out)')
# do another query to get back the result
result = _db.session.execute('SELECT @out').fetchone()
그result
값이 out 매개 변수인 튜플이 됩니다.세션 중에 다른 쿼리가 실패하면 프로시저 호출도 중단(롤백)됩니다.
생성된 프로시저 개체를 실행합니다.func
:
from sqlalchemy import create_engine, func
from sqlalchemy.orm import sessionmaker
engine = create_engine('sqlite://', echo=True)
print engine.execute(func.upper('abc')).scalar() # Using engine
session = sessionmaker(bind=engine)()
print session.execute(func.upper('abc')).scalar() # Using session
SQLAlchemy를 사용하여 MySQL에서 저장 프로시저를 호출하는 가장 쉬운 방법은callproc
의 방법Engine.raw_connection()
.call_proc
호출할 저장 프로시저에 필요한 프로시저 이름과 매개 변수가 필요합니다.
def call_procedure(function_name, params):
connection = cloudsql.Engine.raw_connection()
try:
cursor = connection.cursor()
cursor.callproc(function_name, params)
results = list(cursor.fetchall())
cursor.close()
connection.commit()
return results
finally:
connection.close()
sessionmaker()를 사용하여 세션을 이미 만든 경우 다음 기능을 사용할 수 있습니다.
def exec_procedure(session, proc_name, params):
sql_params = ",".join(["@{0}={1}".format(name, value) for name, value in params.items()])
sql_string = """
DECLARE @return_value int;
EXEC @return_value = [dbo].[{proc_name}] {params};
SELECT 'Return Value' = @return_value;
""".format(proc_name=proc_name, params=sql_params)
return session.execute(sql_string).fetchall()
이제 다음과 같은 매개 변수를 사용하여 저장 프로시저 'MyProc'를 실행할 수 있습니다.
params = {
'Foo': foo_value,
'Bar': bar_value
}
exec_procedure(session, 'MyProc', params)
프로젝트가 절실히 필요해서 저장 프로시저 호출을 처리하는 기능을 작성했습니다.
여기 있습니다.
import sqlalchemy as sql
def execute_db_store_procedure(database, types, sql_store_procedure, *sp_args):
""" Execute the store procedure and return the response table.
Attention: No injection checking!!!
Does work with the CALL syntax as of yet (TODO: other databases).
Attributes:
database -- the database
types -- tuple of strings of SQLAlchemy type names.
Each type describes the type of the argument
with the same number.
List: http://docs.sqlalchemy.org/en/rel_0_7/core/types.html
sql_store_procudure -- string of the stored procedure to be executed
sp_args -- arguments passed to the stored procedure
"""
if not len(types) == len(sp_args):
raise ValueError("types tuple must be the length of the sp args.")
# Construch the type list for the given types
# See
# http://docs.sqlalchemy.org/en/latest/core/sqlelement.html?highlight=expression.text#sqlalchemy.sql.expression.text
# sp_args (and their types) are numbered from 0 to len(sp_args)-1
type_list = [sql.sql.expression.bindparam(
str(no), type_=getattr(sql.types, typ)())
for no, typ in zip(range(len(types)), types)]
try:
# Adapts to the number of arguments given to the function
sp_call = sql.text("CALL `%s`(%s)" % (
sql_store_procedure,
", ".join([":%s" % n for n in range(len(sp_args))])),
bindparams=type_list
)
#raise ValueError("%s\n%s" % (sp_call, type_list))
with database.engine.begin() as connection:
return connection.execute(
sp_call,
# Don't do this at home, kids...
**dict((str(no), arg)
for (no, arg) in zip(range(len(sp_args)), sp_args)))
except sql.exc.DatabaseError:
raise
CALL 구문과 함께 작동하므로 MySQL은 예상대로 작동합니다.MSSQL은 호출 대신 EXEC를 사용하고 구문이 조금 다른 것 같습니다.따라서 서버에 구애받지 않는 상태로 만드는 것은 여러분에게 달려 있지만 그렇게 어렵지는 않을 것입니다.
다른 해결 방법:
query = f'call Procedure ("{@param1}", "{@param2}", "{@param3}")'
sqlEngine = sqlalchemy.create_engine(jdbc)
conn = sqlEngine.connect()
df = pd.read_sql(query,conn,index_col=None)
저는 다음 서명이 있는 postgresql에 대한 저장 프로시저를 가지고 있었습니다.
CREATE OR REPLACE PROCEDURE inc_run_count(
_host text,
_org text,
_repo text,
_rule_ids text[]
)
꽤 많은 오류와 시도 끝에, 저는 이것이 python3에서 프로시저를 호출하는 방법이라는 것을 알게 되었습니다.
def update_db_rule_count(rule_ids: List[str], host: str, org: str, repo: str):
param_dict = {"host": host, "org": org, "repo": repo, "rule_ids": f'{{ {",".join(rule_ids)} }}'}
with sessionmaker(autocommit=False, autoflush=False, bind=ro_engine) as analytics_db:
analytics_db.execute('call inc_run_count(:host, :org, :repo, :rule_ids)', param_dict)
analytics_db.commit()
언급URL : https://stackoverflow.com/questions/3563738/stored-procedures-with-sqlalchemy
'programing' 카테고리의 다른 글
이름을 사용하여 컨트롤을 찾을 수 없습니다. 각 반응 형식의 formControlName (0) | 2023.08.05 |
---|---|
MariaDB help_topic.파일 (0) | 2023.08.05 |
5분마다 새로 고쳐지는 구체화된 보기 만들기 (0) | 2023.08.05 |
jquery ID(공백 포함) (0) | 2023.08.05 |
Axios 400 오류 요청 콜 대신에 호출합니다. (0) | 2023.08.05 |