22 lines
746 B
Python
22 lines
746 B
Python
"""极简 MySQL 访问层(pymysql)。够第1步用,后面要换 SQLAlchemy 也行。"""
|
|
import pymysql
|
|
from pymysql.cursors import DictCursor
|
|
|
|
from .config import settings
|
|
|
|
|
|
def _parse_dsn(dsn: str) -> dict:
|
|
# mysql+pymysql://user:pass@host:port/db?charset=utf8mb4
|
|
body = dsn.split("://", 1)[1]
|
|
creds, rest = body.split("@", 1)
|
|
user, password = creds.split(":", 1)
|
|
hostport, dbq = rest.split("/", 1)
|
|
host, port = (hostport.split(":", 1) + ["3306"])[:2]
|
|
db = dbq.split("?", 1)[0]
|
|
return dict(host=host, port=int(port), user=user, password=password, database=db)
|
|
|
|
|
|
def get_conn():
|
|
cfg = _parse_dsn(settings.MYSQL_DSN)
|
|
return pymysql.connect(charset="utf8mb4", cursorclass=DictCursor, autocommit=True, **cfg)
|