codekingpro/portable-devtools

1# dialects/mysql/asyncmy.py
2# Copyright (C) 2005-2024 the SQLAlchemy authors and contributors <see AUTHORS
3# file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: https://www.opensource.org/licenses/mit-license.php
7# mypy: ignore-errors
8
9r"""
10.. dialect:: mysql+asyncmy
11    :name: asyncmy
12    :dbapi: asyncmy
13    :connectstring: mysql+asyncmy://user:password@host:port/dbname[?key=value&key=value...]
14    :url: https://github.com/long2ice/asyncmy
15
16Using a special asyncio mediation layer, the asyncmy dialect is usable
17as the backend for the :ref:`SQLAlchemy asyncio <asyncio_toplevel>`
18extension package.
19
20This dialect should normally be used only with the
21:func:`_asyncio.create_async_engine` engine creation function::
22
23    from sqlalchemy.ext.asyncio import create_async_engine
24    engine = create_async_engine("mysql+asyncmy://user:pass@hostname/dbname?charset=utf8mb4")
25
26
27"""  # noqa
28from contextlib import asynccontextmanager
29
30from .pymysql import MySQLDialect_pymysql
31from ... import pool
32from ... import util
33from ...engine import AdaptedConnection
34from ...util.concurrency import asyncio
35from ...util.concurrency import await_fallback
36from ...util.concurrency import await_only
37
38
39class AsyncAdapt_asyncmy_cursor:
40    # TODO: base on connectors/asyncio.py
41    # see #10415
42    server_side = False
43    __slots__ = (
44        "_adapt_connection",
45        "_connection",
46        "await_",
47        "_cursor",
48        "_rows",
49    )
50
51    def __init__(self, adapt_connection):
52        self._adapt_connection = adapt_connection
53        self._connection = adapt_connection._connection
54        self.await_ = adapt_connection.await_
55
56        cursor = self._connection.cursor()
57
58        self._cursor = self.await_(cursor.__aenter__())
59        self._rows = []
60
61    @property
62    def description(self):
63        return self._cursor.description
64
65    @property
66    def rowcount(self):
67        return self._cursor.rowcount
68
69    @property
70    def arraysize(self):
71        return self._cursor.arraysize
72
73    @arraysize.setter
74    def arraysize(self, value):
75        self._cursor.arraysize = value
76
77    @property
78    def lastrowid(self):
79        return self._cursor.lastrowid
80
81    def close(self):
82        # note we aren't actually closing the cursor here,
83        # we are just letting GC do it.   to allow this to be async
84        # we would need the Result to change how it does "Safe close cursor".
85        # MySQL "cursors" don't actually have state to be "closed" besides
86        # exhausting rows, which we already have done for sync cursor.
87        # another option would be to emulate aiosqlite dialect and assign
88        # cursor only if we are doing server side cursor operation.
89        self._rows[:] = []
90
91    def execute(self, operation, parameters=None):
92        return self.await_(self._execute_async(operation, parameters))
93
94    def executemany(self, operation, seq_of_parameters):
95        return self.await_(
96            self._executemany_async(operation, seq_of_parameters)
97        )
98
99    async def _execute_async(self, operation, parameters):
100        async with self._adapt_connection._mutex_and_adapt_errors():
101            if parameters is None:
102                result = await self._cursor.execute(operation)
103            else:
104                result = await self._cursor.execute(operation, parameters)
105
106            if not self.server_side:
107                # asyncmy has a "fake" async result, so we have to pull it out
108                # of that here since our default result is not async.
109                # we could just as easily grab "_rows" here and be done with it
110                # but this is safer.
111                self._rows = list(await self._cursor.fetchall())
112            return result
113
114    async def _executemany_async(self, operation, seq_of_parameters):
115        async with self._adapt_connection._mutex_and_adapt_errors():
116            return await self._cursor.executemany(operation, seq_of_parameters)
117
118    def setinputsizes(self, *inputsizes):
119        pass
120
121    def __iter__(self):
122        while self._rows:
123            yield self._rows.pop(0)
124
125    def fetchone(self):
126        if self._rows:
127            return self._rows.pop(0)
128        else:
129            return None
130
131    def fetchmany(self, size=None):
132        if size is None:
133            size = self.arraysize
134
135        retval = self._rows[0:size]
136        self._rows[:] = self._rows[size:]
137        return retval
138
139    def fetchall(self):
140        retval = self._rows[:]
141        self._rows[:] = []
142        return retval
143
144
145class AsyncAdapt_asyncmy_ss_cursor(AsyncAdapt_asyncmy_cursor):
146    # TODO: base on connectors/asyncio.py
147    # see #10415
148    __slots__ = ()
149    server_side = True
150
151    def __init__(self, adapt_connection):
152        self._adapt_connection = adapt_connection
153        self._connection = adapt_connection._connection
154        self.await_ = adapt_connection.await_
155
156        cursor = self._connection.cursor(
157            adapt_connection.dbapi.asyncmy.cursors.SSCursor
158        )
159
160        self._cursor = self.await_(cursor.__aenter__())
161
162    def close(self):
163        if self._cursor is not None:
164            self.await_(self._cursor.close())
165            self._cursor = None
166
167    def fetchone(self):
168        return self.await_(self._cursor.fetchone())
169
170    def fetchmany(self, size=None):
171        return self.await_(self._cursor.fetchmany(size=size))
172
173    def fetchall(self):
174        return self.await_(self._cursor.fetchall())
175
176
177class AsyncAdapt_asyncmy_connection(AdaptedConnection):
178    # TODO: base on connectors/asyncio.py
179    # see #10415
180    await_ = staticmethod(await_only)
181    __slots__ = ("dbapi", "_execute_mutex")
182
183    def __init__(self, dbapi, connection):
184        self.dbapi = dbapi
185        self._connection = connection
186        self._execute_mutex = asyncio.Lock()
187
188    @asynccontextmanager
189    async def _mutex_and_adapt_errors(self):
190        async with self._execute_mutex:
191            try:
192                yield
193            except AttributeError:
194                raise self.dbapi.InternalError(
195                    "network operation failed due to asyncmy attribute error"
196                )
197
198    def ping(self, reconnect):
199        assert not reconnect
200        return self.await_(self._do_ping())
201
202    async def _do_ping(self):
203        async with self._mutex_and_adapt_errors():
204            return await self._connection.ping(False)
205
206    def character_set_name(self):
207        return self._connection.character_set_name()
208
209    def autocommit(self, value):
210        self.await_(self._connection.autocommit(value))
211
212    def cursor(self, server_side=False):
213        if server_side:
214            return AsyncAdapt_asyncmy_ss_cursor(self)
215        else:
216            return AsyncAdapt_asyncmy_cursor(self)
217
218    def rollback(self):
219        self.await_(self._connection.rollback())
220
221    def commit(self):
222        self.await_(self._connection.commit())
223
224    def terminate(self):
225        # it's not awaitable.
226        self._connection.close()
227
228    def close(self) -> None:
229        self.await_(self._connection.ensure_closed())
230
231
232class AsyncAdaptFallback_asyncmy_connection(AsyncAdapt_asyncmy_connection):
233    __slots__ = ()
234
235    await_ = staticmethod(await_fallback)
236
237
238def _Binary(x):
239    """Return x as a binary type."""
240    return bytes(x)
241
242
243class AsyncAdapt_asyncmy_dbapi:
244    def __init__(self, asyncmy):
245        self.asyncmy = asyncmy
246        self.paramstyle = "format"
247        self._init_dbapi_attributes()
248
249    def _init_dbapi_attributes(self):
250        for name in (
251            "Warning",
252            "Error",
253            "InterfaceError",
254            "DataError",
255            "DatabaseError",
256            "OperationalError",
257            "InterfaceError",
258            "IntegrityError",
259            "ProgrammingError",
260            "InternalError",
261            "NotSupportedError",
262        ):
263            setattr(self, name, getattr(self.asyncmy.errors, name))
264
265    STRING = util.symbol("STRING")
266    NUMBER = util.symbol("NUMBER")
267    BINARY = util.symbol("BINARY")
268    DATETIME = util.symbol("DATETIME")
269    TIMESTAMP = util.symbol("TIMESTAMP")
270    Binary = staticmethod(_Binary)
271
272    def connect(self, *arg, **kw):
273        async_fallback = kw.pop("async_fallback", False)
274        creator_fn = kw.pop("async_creator_fn", self.asyncmy.connect)
275
276        if util.asbool(async_fallback):
277            return AsyncAdaptFallback_asyncmy_connection(
278                self,
279                await_fallback(creator_fn(*arg, **kw)),
280            )
281        else:
282            return AsyncAdapt_asyncmy_connection(
283                self,
284                await_only(creator_fn(*arg, **kw)),
285            )
286
287
288class MySQLDialect_asyncmy(MySQLDialect_pymysql):
289    driver = "asyncmy"
290    supports_statement_cache = True
291
292    supports_server_side_cursors = True
293    _sscursor = AsyncAdapt_asyncmy_ss_cursor
294
295    is_async = True
296    has_terminate = True
297
298    @classmethod
299    def import_dbapi(cls):
300        return AsyncAdapt_asyncmy_dbapi(__import__("asyncmy"))
301
302    @classmethod
303    def get_pool_class(cls, url):
304        async_fallback = url.query.get("async_fallback", False)
305
306        if util.asbool(async_fallback):
307            return pool.FallbackAsyncAdaptedQueuePool
308        else:
309            return pool.AsyncAdaptedQueuePool
310
311    def do_terminate(self, dbapi_connection) -> None:
312        dbapi_connection.terminate()
313
314    def create_connect_args(self, url):
315        return super().create_connect_args(
316            url, _translate_args=dict(username="user", database="db")
317        )
318
319    def is_disconnect(self, e, connection, cursor):
320        if super().is_disconnect(e, connection, cursor):
321            return True
322        else:
323            str_e = str(e).lower()
324            return (
325                "not connected" in str_e or "network operation failed" in str_e
326            )
327
328    def _found_rows_client_flag(self):
329        from asyncmy.constants import CLIENT
330
331        return CLIENT.FOUND_ROWS
332
333    def get_driver_connection(self, connection):
334        return connection._connection
335
336
337dialect = MySQLDialect_asyncmy
338