45 lines
1.5 KiB
Python
45 lines
1.5 KiB
Python
# app/repositories/queries/base_queries.py
|
|
"""
|
|
HealthSync AI 기본 시스템 쿼리 모음
|
|
"""
|
|
|
|
|
|
class BaseQueries:
|
|
"""기본 시스템 쿼리"""
|
|
|
|
# 데이터베이스 연결 테스트
|
|
CONNECTION_TEST = "SELECT 1"
|
|
|
|
DATABASE_VERSION = "SELECT version()"
|
|
|
|
CURRENT_DATABASE = "SELECT current_database()"
|
|
|
|
CURRENT_USER = "SELECT current_user"
|
|
|
|
# 테이블 목록 조회
|
|
LIST_TABLES = """
|
|
SELECT table_name,
|
|
table_schema,
|
|
table_type
|
|
FROM information_schema.tables
|
|
WHERE table_schema NOT IN ('information_schema', 'pg_catalog')
|
|
ORDER BY table_schema, table_name LIMIT 20 \
|
|
"""
|
|
|
|
# 테이블 컬럼 정보 조회
|
|
GET_TABLE_COLUMNS = """
|
|
SELECT column_name
|
|
FROM information_schema.columns
|
|
WHERE table_name = :table_name
|
|
ORDER BY ordinal_position \
|
|
"""
|
|
|
|
# 테이블 데이터 조회 (동적 쿼리 - 주의해서 사용)
|
|
@staticmethod
|
|
def get_table_data_query(table_name: str, limit: int = 5) -> str:
|
|
"""테이블 데이터 조회 쿼리 생성 (SQL 인젝션 방지를 위한 검증 필요)"""
|
|
# 테이블 이름 검증
|
|
if not table_name.replace('_', '').replace('-', '').isalnum():
|
|
raise ValueError("잘못된 테이블 이름입니다.")
|
|
|
|
return f"SELECT * FROM {table_name} LIMIT {limit}" |