?HAMA

Navigation

Breadcrumbs

/ / opt / hc_python / lib64 / python3.12 / site-packages / sqlalchemy / dialects / postgresql

Create

Create File | Create Folder | Upload Files | Run Command

Edit File Size: 5.56 KB

//opt/hc_python/lib64/python3.12/site-packages/sqlalchemy/dialects/postgresql/_psycopg_common.py

Text Base64

# dialects/postgresql/_psycopg_common.py
# Copyright (C) 2005-2025 the SQLAlchemy authors and contributors
# <see AUTHORS file>
#
# This module is part of SQLAlchemy and is released under
# the MIT License: https://www.opensource.org/licenses/mit-license.php
# mypy: ignore-errors
from __future__ import annotations

import decimal

from .array import ARRAY as PGARRAY
from .base import _DECIMAL_TYPES
from .base import _FLOAT_TYPES
from .base import _INT_TYPES
from .base import PGDialect
from .base import PGExecutionContext
from .hstore import HSTORE
from .pg_catalog import _SpaceVector
from .pg_catalog import INT2VECTOR
from .pg_catalog import OIDVECTOR
from ... import exc
from ... import types as sqltypes
from ... import util
from ...engine import processors

_server_side_id = util.counter()

class _PsycopgNumeric(sqltypes.Numeric):
    def bind_processor(self, dialect):
        return None

def result_processor(self, dialect, coltype):
        if self.asdecimal:
            if coltype in _FLOAT_TYPES:
                return processors.to_decimal_processor_factory(
                    decimal.Decimal, self._effective_decimal_return_scale
                )
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                # psycopg returns Decimal natively for 1700
                return None
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype
                )
        else:
            if coltype in _FLOAT_TYPES:
                # psycopg returns float natively for 701
                return None
            elif coltype in _DECIMAL_TYPES or coltype in _INT_TYPES:
                return processors.to_float
            else:
                raise exc.InvalidRequestError(
                    "Unknown PG numeric type: %d" % coltype
                )

class _PsycopgFloat(_PsycopgNumeric):
    __visit_name__ = "float"

class _PsycopgHStore(HSTORE):
    def bind_processor(self, dialect):
        if dialect._has_native_hstore:
            return None
        else:
            return super().bind_processor(dialect)

def result_processor(self, dialect, coltype):
        if dialect._has_native_hstore:
            return None
        else:
            return super().result_processor(dialect, coltype)

class _PsycopgARRAY(PGARRAY):
    render_bind_cast = True

class _PsycopgINT2VECTOR(_SpaceVector, INT2VECTOR):
    pass

class _PsycopgOIDVECTOR(_SpaceVector, OIDVECTOR):
    pass

class _PGExecutionContext_common_psycopg(PGExecutionContext):
    def create_server_side_cursor(self):
        # use server-side cursors:
        # psycopg
        # https://www.psycopg.org/psycopg3/docs/advanced/cursors.html#server-side-cursors
        # psycopg2
        # https://www.psycopg.org/docs/usage.html#server-side-cursors
        ident = "c_%s_%s" % (hex(id(self))[2:], hex(_server_side_id())[2:])
        return self._dbapi_connection.cursor(ident)

class _PGDialect_common_psycopg(PGDialect):
    supports_statement_cache = True
    supports_server_side_cursors = True

default_paramstyle = "pyformat"

_has_native_hstore = True

colspecs = util.update_copy(
        PGDialect.colspecs,
        {
            sqltypes.Numeric: _PsycopgNumeric,
            sqltypes.Float: _PsycopgFloat,
            HSTORE: _PsycopgHStore,
            sqltypes.ARRAY: _PsycopgARRAY,
            INT2VECTOR: _PsycopgINT2VECTOR,
            OIDVECTOR: _PsycopgOIDVECTOR,
        },
    )

def __init__(
        self,
        client_encoding=None,
        use_native_hstore=True,
        **kwargs,
    ):
        PGDialect.__init__(self, **kwargs)
        if not use_native_hstore:
            self._has_native_hstore = False
        self.use_native_hstore = use_native_hstore
        self.client_encoding = client_encoding

def create_connect_args(self, url):
        opts = url.translate_connect_args(username="user", database="dbname")

multihosts, multiports = self._split_multihost_from_url(url)

if opts or url.query:
            if not opts:
                opts = {}
            if "port" in opts:
                opts["port"] = int(opts["port"])
            opts.update(url.query)

if multihosts:
                opts["host"] = ",".join(multihosts)
                comma_ports = ",".join(str(p) if p else "" for p in multiports)
                if comma_ports:
                    opts["port"] = comma_ports
            return ([], opts)
        else:
            # no connection arguments whatsoever; psycopg2.connect()
            # requires that "dsn" be present as a blank string.
            return ([""], opts)

def get_isolation_level_values(self, dbapi_connection):
        return (
            "AUTOCOMMIT",
            "READ COMMITTED",
            "READ UNCOMMITTED",
            "REPEATABLE READ",
            "SERIALIZABLE",
        )

def set_deferrable(self, connection, value):
        connection.deferrable = value

def get_deferrable(self, connection):
        return connection.deferrable

def _do_autocommit(self, connection, value):
        connection.autocommit = value

def do_ping(self, dbapi_connection):
        cursor = None
        before_autocommit = dbapi_connection.autocommit

if not before_autocommit:
            dbapi_connection.autocommit = True
        cursor = dbapi_connection.cursor()
        try:
            cursor.execute(self._dialect_specific_select_one)
        finally:
            cursor.close()
            if not before_autocommit and not dbapi_connection.closed:
                dbapi_connection.autocommit = before_autocommit

return True

Exit & Reset Text mode: syntax highlighting auto-detects file type.

Directory Contents

Dirs: 1 × Files: 19

Sort:

Enable drag-to-move

Name	Size	Perms	Modified	Actions
__pycache__ DIR	-	drwxr-xr-x	2025-04-04 08:02:01	Edit Download recursive recursive
array.py	13.53 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
asyncpg.py	40.32 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
base.py	175.46 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
dml.py	11.84 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
ext.py	15.94 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
hstore.py	11.65 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
json.py	12.57 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
named_types.py	17.16 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
operators.py	2.74 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
pg8000.py	18.20 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
pg_catalog.py	9.04 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
provision.py	5.63 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
psycopg.py	22.78 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
psycopg2.py	31.28 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
psycopg2cffi.py	1.71 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
ranges.py	32.21 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
types.py	7.45 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
_psycopg_common.py	5.56 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive
__init__.py	3.80 KB	lrw-r--r--	2025-04-04 08:02:01	Edit Download recursive recursive