Class PGDialect_psycopg2

lib/sqlalchemy/dialects/postgresql/psycopg2.py:605–858

class PGDialect_psycopg2(_PGDialect_common_psycopg):
    driver = "psycopg2"

    supports_statement_cache = True
    supports_server_side_cursors = True

    default_paramstyle = "pyformat"
    # set to true based on psycopg2 version
    supports_sane_multi_rowcount = False
    execution_ctx_cls = PGExecutionContext_psycopg2
    preparer = PGIdentifierPreparer_psycopg2
    psycopg2_version = (0, 0)
    use_insertmanyvalues_wo_returning = True

    returns_native_bytes = False

    _has_native_hstore = True

    colspecs = util.update_copy(
        _PGDialect_common_psycopg.colspecs,
        {
            JSON: _PGJSON,
            sqltypes.JSON: _PGJSON,
            JSONB: _PGJSONB,
            ranges.INT4RANGE: _Psycopg2NumericRange,
            ranges.INT8RANGE: _Psycopg2NumericRange,
            ranges.NUMRANGE: _Psycopg2NumericRange,
            ranges.DATERANGE: _Psycopg2DateRange,
            ranges.TSRANGE: _Psycopg2DateTimeRange,
            ranges.TSTZRANGE: _Psycopg2DateTimeTZRange,
        },
    )

    def __init__(
        self,
        executemany_mode="values_only",
        executemany_batch_page_size=100,
        **kwargs,
    ):
        _PGDialect_common_psycopg.__init__(self, **kwargs)

        if self._native_inet_types:
            raise NotImplementedError(
                "The psycopg2 dialect does not implement "
                "ipaddress type handling; native_inet_types cannot be set "
                "to ``True`` when using this dialect."
            )

        # Parse executemany_mode argument, allowing it to be only one of the
        # symbol names
        self.executemany_mode = parse_user_argument_for_enum(
            executemany_mode,
            {
                EXECUTEMANY_VALUES: ["values_only"],
                EXECUTEMANY_VALUES_PLUS_BATCH: ["values_plus_batch"],
            },
            "executemany_mode",
        )