다중DB + 커넥션풀캐시 + 데코레이터

다중DB 비동기커넥션풀캐시 데코레이터

여러 DB 선택 처리를 지원하도록 비동기 커넥션 풀 구조
다수의 DB가 db_config.json에 정의
데코레이터에서 DB 이름만 바꾸면 자동으로 해당 DB에 접속되도록 구현

db_config.json 예시 (여러 DB)

{
  "Database": [
    {
      "Name": "MyDB1",
      "Username": "user1",
      "Password": "pass1",
      "Host": "localhost",
      "Port": 1521,
      "ServiceName": "orclpdb1",
      "Min": 2,
      "Max": 5,
      "Increment": 1
    },
    {
      "Name": "MyDB2",
      "Username": "user2",
      "Password": "pass2",
      "Host": "remotehost",
      "Port": 1521,
      "ServiceName": "orclpdb2",
      "Min": 1,
      "Max": 3,
      "Increment": 1
    }
  ]
}

여러 DB 자동 선택 + 커넥션 풀 캐시 + 데코레이터

import oracledb
import asyncio
import json
from functools import wraps

# 전역 커넥션 풀 저장소
connection_pools = {}

# DB 설정 불러오기
def load_db_config(name):
    with open("db_config.json", "r") as f:
        config = json.load(f)
    for db in config["Database"]:
        if db["Name"] == name:
            return db
    raise ValueError(f"No DB config found for name: {name}")

# DB 이름 기반 커넥션 풀 초기화
async def init_connection_pool(name):
    if name in connection_pools:
        return connection_pools[name]

    cfg = load_db_config(name)

    pool = await oracledb.create_pool(
        user=cfg["Username"],
        password=cfg["Password"],
        host=cfg["Host"],
        port=cfg["Port"],
        service_name=cfg["ServiceName"],
        min=cfg["Min"],
        max=cfg["Max"],
        increment=cfg["Increment"],
        getmode=oracledb.SPOOL_ATTRVAL_WAIT,
        timeout=10
    )
    connection_pools[name] = pool
    return pool

# 비동기 데코레이터 - DB 선택 가능
def with_async_connection(db_name):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            pool = await init_connection_pool(db_name)
            async with pool.acquire() as conn:
                async with conn.cursor() as cursor:
                    return await func(cursor, *args, **kwargs)
        return wrapper
    return decorator

사용 예시: DB별 쿼리

@with_async_connection("MyDB1")
async def get_users_from_db1(cursor):
    await cursor.execute("SELECT * FROM users")
    return await cursor.fetchall()

@with_async_connection("MyDB2")
async def get_products_from_db2(cursor):
    await cursor.execute("SELECT * FROM products")
    return await cursor.fetchall()

async def main():
    users = await get_users_from_db1()
    products = await get_products_from_db2()
    print("Users:", users)
    print("Products:", products)

if __name__ == "__main__":
    asyncio.run(main())

제목을넣으세요

추가기능
- @with_async_connection(db_name_func()) → 런타임에 동적으로 DB 선택

Comments