SQLAlchemy를 사용하여 런타임에 사용자 정의 SQL 쿼리를 실행하는 방법은 무엇입니까?

Question

은 내가 솔루션을 식별하기 위해 노력하고있는 유스 케이스 구현을 가지고, 구현은 다음과 같다 :

• 사용자가 MySQL을 같이 (DATABASE_TYPE 같은 SQL 서버 연결 매개 변수를 정의를 pgsql, sqlite가 또는 DB 지원 SQLAlchemy의)에 의해, HOST, PORT, SCHEMA_NAME, USER_NAME 및 PASSWORD는 사용 후 제공되는 연결 매개 변수를 확인하기 위해 연결 테스트가 수행하고,
성공
• 는, 사용자가 SQL 쿼리를 입력합니다 성공과 실패 피드백을 제공 (
• 다음 SQLAlchemy의 SQL 표현 언어에 대한 자세한 내용은 here) 에드.

위에서 언급 한 모든 SQLAlchemy 매개 변수가 런타임에 사용자 정의됩니다. 위의 연결 매개 변수와 SQL 쿼리를 사용하는 Python에서이 함수를 구현하려고합니다. 성공적인 연결에 대한 쿼리를 실행하고 쿼리 결과를 반환합니다. 이것에 대한 방향이나 도움이된다면 도움이 될 것입니다.

덕분에 지금까지 만 SQLite는과 포스트 그레스에 대한

Answer 1

나는 큰 프로젝트의 일환으로 이런 일을 보내고있어,하지만이 일반화 될 수 있습니다.

사용자가 DATABASE_TYPE (MySQL, pgsql, sqlite 또는 SQLAlchemy에서 지원하는 DB), HOST, PORT, SCHEMA_NAME, USER_NAME 및 PASSWORD를 알기를 원한다면 DB URL을 묻는 것이 좋습니다. 조금 더 쉬워.) (

from sqlalchemy import create_engine 
    try: 
     engine = create_engine(url)   
     connection = engine.connect() 
     connection.close() 
    except Exception, e: 
     LOGGER.exception(e) 
     raise MyCusomException('Could not initialize db. Invalid URL.') 

의 맞춤 SQL 쿼리 텍스트를 사용하여 실행할 수 있습니다 here를 같이 구성 :

그 후에는하여 연결을 테스트 할 수 있습니다.