Zy Filemanager

📁 File Browser

Name	Size	Permissions	Actions
📁 __pycache__	DIR	0755	🗑️
📄 __init__.py (1,483 bytes)	1.45 KB	0644	✍️ ⬇️ 🗑️
📄 base.py (36,523 bytes)	35.67 KB	0644	✍️ ⬇️ 🗑️
📄 dbapi_proxy.py (4,320 bytes)	4.22 KB	0644	✍️ ⬇️ 🗑️
📄 impl.py (14,958 bytes)	14.61 KB	0644	✍️ ⬇️ 🗑️

📤 Upload Files

Single File

Multiple Files

ZIP Extract

ZIP file will be deleted after extraction

➕ Create New

New Folder

New File

✍️ Editing: dbapi_proxy.py

            File: /opt/cloudlinux/venv/lib/python3.11/site-packages/sqlalchemy/pool/dbapi_proxy.py

            Size: 4.22 KB | 
            Last Modified: 2026-01-20 20:01:47        

# sqlalchemy/pool/dbapi_proxy.py
# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: http://www.opensource.org/licenses/mit-license.php

"""DBAPI proxy utility.

Provides transparent connection pooling on top of a Python DBAPI.

This is legacy SQLAlchemy functionality that is not typically used
today.

"""

from .impl import QueuePool
from .. import util
from ..util import threading

proxies = {}

@util.deprecated(
    "1.3",
    "The :func:`.pool.manage` function is deprecated, and will be "
    "removed in a future release.",
)
def manage(module, **params):
    r"""Return a proxy for a DB-API module that automatically
    pools connections.

Given a DB-API 2.0 module and pool management parameters, returns
    a proxy for the module that will automatically pool connections,
    creating new connection pools for each distinct set of connection
    arguments sent to the decorated module's connect() function.

:param module: a DB-API 2.0 database module

:param poolclass: the class used by the pool module to provide
      pooling.  Defaults to :class:`.QueuePool`.

:param \**params: will be passed through to *poolclass*

"""
    try:
        return proxies[module]
    except KeyError:
        return proxies.setdefault(module, _DBProxy(module, **params))

def clear_managers():
    """Remove all current DB-API 2.0 managers.

All pools and connections are disposed.
    """

for manager in proxies.values():
        manager.close()
    proxies.clear()

class _DBProxy(object):

"""Layers connection pooling behavior on top of a standard DB-API module.

Proxies a DB-API 2.0 connect() call to a connection pool keyed to the
    specific connect parameters. Other functions and attributes are delegated
    to the underlying DB-API module.
    """

def __init__(self, module, poolclass=QueuePool, **kw):
        """Initializes a new proxy.

module
          a DB-API 2.0 module

poolclass
          a Pool class, defaulting to QueuePool

Other parameters are sent to the Pool object's constructor.

"""

self.module = module
        self.kw = kw
        self.poolclass = poolclass
        self.pools = {}
        self._create_pool_mutex = threading.Lock()

def close(self):
        for key in list(self.pools):
            del self.pools[key]

def __del__(self):
        self.close()

def __getattr__(self, key):
        return getattr(self.module, key)

def get_pool(self, *args, **kw):
        key = self._serialize(*args, **kw)
        try:
            return self.pools[key]
        except KeyError:
            self._create_pool_mutex.acquire()
            try:
                if key not in self.pools:
                    kw.pop("sa_pool_key", None)
                    pool = self.poolclass(
                        lambda: self.module.connect(*args, **kw), **self.kw
                    )
                    self.pools[key] = pool
                    return pool
                else:
                    return self.pools[key]
            finally:
                self._create_pool_mutex.release()

def connect(self, *args, **kw):
        """Activate a connection to the database.

Connect to the database using this DBProxy's module and the given
        connect arguments.  If the arguments match an existing pool, the
        connection will be returned from the pool's current thread-local
        connection instance, or if there is no thread-local connection
        instance it will be checked out from the set of pooled connections.

If the pool has no available connections and allows new connections
        to be created, a new database connection will be made.

"""

return self.get_pool(*args, **kw).connect()

def dispose(self, *args, **kw):
        """Dispose the pool referenced by the given connect arguments."""

key = self._serialize(*args, **kw)
        try:
            del self.pools[key]
        except KeyError:
            pass

def _serialize(self, *args, **kw):
        if "sa_pool_key" in kw:
            return kw["sa_pool_key"]

return tuple(list(args) + [(k, kw[k]) for k in sorted(kw)])

↩️ Back