κ°λ°/DB
DBμ μ TEST CODE(python)
-=HaeJuK=-
2025. 5. 13. 15:20
728x90
λ°μν
π DB μ μ ν μ€νΈ λꡬ μμ½
π§© ν μ€νΈ λμ
- MariaDB (ν¬νΈ 3306)
- Microsoft SQL Server (ν¬νΈ 1433)
π§ ν μ€νΈ νλͺ©
ν μ€νΈ μ’ λ₯ | μ€λͺ |
---|---|
test1 |
DB μ μ ν μ¦μ μ’ λ£ (μ°κ²°λ§ νμΈ) |
test2 |
DB μ μ β νλμ½λ©λ 쿼리 μ€ν β μ’ λ£ |
π§ͺ νλμ½λ©λ 쿼리 (test2)
DB νμ | 쿼리 |
---|---|
MariaDB | SELECT NOW() |
MSSQL | SELECT GETDATE() |
π λͺ λ Ήμ΄ μμ
# test1 - μ μλ§ ν
μ€νΈ (10ν λ°λ³΅)
python test_db_loop.py test1 mariadb 127.0.0.1 3306 root mypass testdb 10
# test2 - 쿼리 ν
μ€νΈ ν¬ν¨ (5ν λ°λ³΅)
python test_db_loop.py test2 mssql 127.0.0.1 1433 sa pass123 mydb 5
π§΅ λ³λ ¬ μ€ν μμ (λ©ν° νλ‘μΈμ€)
from multiprocessing import Process
from test_db_lib import test1_connect_only
def run_task():
test1_connect_only(
dbtype="mariadb",
host="192.168.0.10",
port="3306",
user="root",
password="mypassword",
dbname="test"
)
if __name__ == "__main__":
processes = [Process(target=run_task) for _ in range(10)]
for p in processes:
p.start()
for p in processes:
p.join()
π§ μ£Όμ Python μ½λ κ΅¬μ± (test1/test2)
def test1_connect_only(dbtype, host, port, user, password, dbname):
if dbtype == "mariadb":
import pymysql
conn = pymysql.connect(
host=host, port=int(port), user=user, password=password,
database=dbname, connect_timeout=5
)
conn.close()
print("[OK] μ μ/μ’
λ£ μ±κ³΅")
return True
elif dbtype == "mssql":
import pyodbc
conn_str = (
f"DRIVER={{ODBC Driver 17 for SQL Server}};"
f"SERVER={host},{port};DATABASE={dbname};UID={user};PWD={password};"
f"TrustServerCertificate=yes;"
)
conn = pyodbc.connect(conn_str, timeout=5)
conn.close()
print("[OK] μ μ/μ’
λ£ μ±κ³΅")
return True
β οΈ μ μ μ¬ν
- Windowsμμλ λ°λμ
__name__ == "__main__"
쑰건문 μμμ νλ‘μΈμ€ μ€ν - DB κ³μ λμ μ μ μ ν, 컀λ₯μ ν μ΅λκ° λ± νμΈ νμ
- 쿼리 μ€ν¨ λλ λ€νΈμν¬ μλ¬λ λ‘κ·Έλ‘ μΆλ ₯λ¨
728x90