Devcafe: 새 문서: == 다중DB 비동기커넥션풀캐시 데코레이터 == * 여러 DB 선택 처리를 지원하도록 비동기 커넥션 풀 구조 * 다수의 DB가 db_config.json에 정의 * 데코레이터에서 DB 이름만 바꾸면 자동으로 해당 DB에 접속되도록 구현 === db_config.json 예시 (여러 DB) === { "Database": [ { "Name": "MyDB1", "Username": "user1", "Password": "pass1", "Host": "localhost", "Port": 1521,...

2025-06-16T23:35:39Z

새 문서: == 다중DB 비동기커넥션풀캐시 데코레이터 == * 여러 DB 선택 처리를 지원하도록 비동기 커넥션 풀 구조 * 다수의 DB가 db_config.json에 정의 * 데코레이터에서 DB 이름만 바꾸면 자동으로 해당 DB에 접속되도록 구현 === db_config.json 예시 (여러 DB) === <source lang=js> { "Database": [ { "Name": "MyDB1", "Username": "user1", "Password": "pass1", "Host": "localhost", "Port": 1521,...

새 문서

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

=== db_config.json 예시 (여러 DB) ===
<source lang=js>
{
"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
}
]
}

</source>
----

=== 여러 DB 자동 선택 + 커넥션 풀 캐시 + 데코레이터 ===
<source lang=python>
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
</source>

=== 사용 예시: DB별 쿼리 ===
<source lang=python>
@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())
</source>

{{요약
|내용=기능설명
# 다중 DB 설정
# JSON 파일에서 DB별로 설정
# DB 이름에 따라 자동 연결
# @with_async_connection("MyDB1") 방식
# 커넥션 풀 재사용
## 이미 만들어진 풀은 재사용 (성능 최적화)
}}

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

[[category:python]]

다중DB + 커넥션풀캐시 + 데코레이터 - 편집 역사