Sorin Savu
46 lines
1.2 KiB
Python
46 lines
1.2 KiB
Python
"""MySQL connection helper (PyMySQL).
|
|
|
|
Connection settings come from environment variables so nothing is hard-coded:
|
|
|
|
WIKI_DB_HOST default 127.0.0.1
|
|
WIKI_DB_PORT default 3306
|
|
WIKI_DB_USER default root
|
|
WIKI_DB_PASSWORD default "" (empty)
|
|
WIKI_DB_NAME default wiki
|
|
|
|
`connect()` selects the wiki database; `connect(select_db=False)` connects to the
|
|
server without a database (used by migrate.py to CREATE DATABASE).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
|
|
|
|
def _cfg() -> dict:
|
|
return {
|
|
"host": os.environ.get("WIKI_DB_HOST", "127.0.0.1"),
|
|
"port": int(os.environ.get("WIKI_DB_PORT", "3306")),
|
|
"user": os.environ.get("WIKI_DB_USER", "root"),
|
|
"password": os.environ.get("WIKI_DB_PASSWORD", ""),
|
|
}
|
|
|
|
|
|
def db_name() -> str:
|
|
return os.environ.get("WIKI_DB_NAME", "wiki")
|
|
|
|
|
|
def connect(select_db: bool = True) -> pymysql.connections.Connection:
|
|
kwargs = _cfg()
|
|
if select_db:
|
|
kwargs["database"] = db_name()
|
|
return pymysql.connect(
|
|
cursorclass=DictCursor,
|
|
autocommit=True,
|
|
charset="utf8mb4",
|
|
**kwargs,
|
|
)
|