-
Notifications
You must be signed in to change notification settings - Fork 59
/
__init__.py
118 lines (87 loc) · 3.9 KB
/
__init__.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
from importlib.metadata import version
flask_major_version = int(version("flask")[0])
import MySQLdb
from MySQLdb import cursors
from flask import current_app
if flask_major_version >= 2:
from flask import g
ctx = g
else:
from flask import _app_ctx_stack
ctx = _app_ctx_stack.top
class MySQL:
def __init__(self, app=None):
self.app = app
if app is not None:
self.init_app(app)
def init_app(self, app):
"""Initialize the `app` for use with this
:class:`~flask_mysqldb.MySQL` class.
This is called automatically if `app` is passed to
:meth:`~MySQL.__init__`.
:param flask.Flask app: the application to configure for use with
this :class:`~flask_mysqldb.MySQL` class.
"""
app.config.setdefault("MYSQL_HOST", "localhost")
app.config.setdefault("MYSQL_USER", None)
app.config.setdefault("MYSQL_PASSWORD", None)
app.config.setdefault("MYSQL_DB", None)
app.config.setdefault("MYSQL_PORT", 3306)
app.config.setdefault("MYSQL_UNIX_SOCKET", None)
app.config.setdefault("MYSQL_CONNECT_TIMEOUT", 10)
app.config.setdefault("MYSQL_READ_DEFAULT_FILE", None)
app.config.setdefault("MYSQL_USE_UNICODE", True)
app.config.setdefault("MYSQL_CHARSET", "utf8")
app.config.setdefault("MYSQL_SQL_MODE", None)
app.config.setdefault("MYSQL_CURSORCLASS", None)
app.config.setdefault("MYSQL_AUTOCOMMIT", False)
app.config.setdefault("MYSQL_CUSTOM_OPTIONS", None)
if hasattr(app, "teardown_appcontext"):
app.teardown_appcontext(self.teardown)
@property
def connect(self):
kwargs = {}
if current_app.config["MYSQL_HOST"]:
kwargs["host"] = current_app.config["MYSQL_HOST"]
if current_app.config["MYSQL_USER"]:
kwargs["user"] = current_app.config["MYSQL_USER"]
if current_app.config["MYSQL_PASSWORD"]:
kwargs["passwd"] = current_app.config["MYSQL_PASSWORD"]
if current_app.config["MYSQL_DB"]:
kwargs["db"] = current_app.config["MYSQL_DB"]
if current_app.config["MYSQL_PORT"]:
kwargs["port"] = current_app.config["MYSQL_PORT"]
if current_app.config["MYSQL_UNIX_SOCKET"]:
kwargs["unix_socket"] = current_app.config["MYSQL_UNIX_SOCKET"]
if current_app.config["MYSQL_CONNECT_TIMEOUT"]:
kwargs["connect_timeout"] = current_app.config["MYSQL_CONNECT_TIMEOUT"]
if current_app.config["MYSQL_READ_DEFAULT_FILE"]:
kwargs["read_default_file"] = current_app.config["MYSQL_READ_DEFAULT_FILE"]
if current_app.config["MYSQL_USE_UNICODE"]:
kwargs["use_unicode"] = current_app.config["MYSQL_USE_UNICODE"]
if current_app.config["MYSQL_CHARSET"]:
kwargs["charset"] = current_app.config["MYSQL_CHARSET"]
if current_app.config["MYSQL_SQL_MODE"]:
kwargs["sql_mode"] = current_app.config["MYSQL_SQL_MODE"]
if current_app.config["MYSQL_CURSORCLASS"]:
kwargs["cursorclass"] = getattr(
cursors, current_app.config["MYSQL_CURSORCLASS"]
)
if current_app.config["MYSQL_AUTOCOMMIT"]:
kwargs["autocommit"] = current_app.config["MYSQL_AUTOCOMMIT"]
if current_app.config["MYSQL_CUSTOM_OPTIONS"]:
kwargs.update(current_app.config["MYSQL_CUSTOM_OPTIONS"])
return MySQLdb.connect(**kwargs)
@property
def connection(self):
"""Attempts to connect to the MySQL server.
:return: Bound MySQL connection object if successful or ``None`` if
unsuccessful.
"""
if ctx is not None:
if not hasattr(ctx, "mysql_db"):
ctx.mysql_db = self.connect
return ctx.mysql_db
def teardown(self, exception):
if hasattr(ctx, "mysql_db"):
ctx.mysql_db.close()