Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/mysql/gaerdbms.py : 38%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# mysql/gaerdbms.py
2# Copyright (C) 2005-2020 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: http://www.opensource.org/licenses/mit-license.php
7r"""
8.. dialect:: mysql+gaerdbms
9 :name: Google Cloud SQL
10 :dbapi: rdbms
11 :connectstring: mysql+gaerdbms:///<dbname>?instance=<instancename>
12 :url: https://developers.google.com/appengine/docs/python/cloud-sql/developers-guide
14 This dialect is based primarily on the :mod:`.mysql.mysqldb` dialect with
15 minimal changes.
17 .. deprecated:: 1.0 This dialect is **no longer necessary** for
18 Google Cloud SQL; the MySQLdb dialect can be used directly.
19 Cloud SQL now recommends creating connections via the
20 mysql dialect using the URL format
22 ``mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/<projectid>:<instancename>``
25Pooling
26-------
28Google App Engine connections appear to be randomly recycled,
29so the dialect does not pool connections. The :class:`.NullPool`
30implementation is installed within the :class:`_engine.Engine` by
31default.
33""" # noqa
35import os
36import re
38from sqlalchemy.util import warn_deprecated
39from .mysqldb import MySQLDialect_mysqldb
40from ...pool import NullPool
43def _is_dev_environment():
44 return os.environ.get("SERVER_SOFTWARE", "").startswith("Development/")
47class MySQLDialect_gaerdbms(MySQLDialect_mysqldb):
48 @classmethod
49 def dbapi(cls):
51 warn_deprecated(
52 "Google Cloud SQL now recommends creating connections via the "
53 "MySQLdb dialect directly, using the URL format "
54 "mysql+mysqldb://root@/<dbname>?unix_socket=/cloudsql/"
55 "<projectid>:<instancename>"
56 )
58 # from django:
59 # http://code.google.com/p/googleappengine/source/
60 # browse/trunk/python/google/storage/speckle/
61 # python/django/backend/base.py#118
62 # see also [ticket:2649]
63 # see also http://stackoverflow.com/q/14224679/34549
64 from google.appengine.api import apiproxy_stub_map
66 if _is_dev_environment():
67 from google.appengine.api import rdbms_mysqldb
69 return rdbms_mysqldb
70 elif apiproxy_stub_map.apiproxy.GetStub("rdbms"):
71 from google.storage.speckle.python.api import rdbms_apiproxy
73 return rdbms_apiproxy
74 else:
75 from google.storage.speckle.python.api import rdbms_googleapi
77 return rdbms_googleapi
79 @classmethod
80 def get_pool_class(cls, url):
81 # Cloud SQL connections die at any moment
82 return NullPool
84 def create_connect_args(self, url):
85 opts = url.translate_connect_args()
86 if not _is_dev_environment():
87 # 'dsn' and 'instance' are because we are skipping
88 # the traditional google.api.rdbms wrapper
89 opts["dsn"] = ""
90 opts["instance"] = url.query["instance"]
91 return [], opts
93 def _extract_error_code(self, exception):
94 match = re.compile(r"^(\d+)L?:|^\((\d+)L?,").match(str(exception))
95 # The rdbms api will wrap then re-raise some types of errors
96 # making this regex return no matches.
97 code = match.group(1) or match.group(2) if match else None
98 if code:
99 return int(code)
102dialect = MySQLDialect_gaerdbms