--- /dev/null
+.. change::
+ :tags: bug, postgresql
+
+ The pg8000 dialect has been revised and modernized for the most recent
+ version of the pg8000 driver for PostgreSQL. Changes to the dialect
+ include:
+
+ * All data types are now sent as text rather than binary.
+
+ * Using adapters, custom types can be plugged in to pg8000.
+
+ * Previously, named prepared statements were used for all statements.
+ Now unnamed prepared statements are used by default, and named
+ prepared statements can be used explicitly by calling the
+ Connection.prepare() method, which returns a PreparedStatement
+ object.
+
+ Pull request courtesy Tony Locke.
@generate_driver_url.for_db("mysql", "mariadb")
-def generate_driver_url(url, driver):
+def generate_driver_url(url, driver, query):
backend = url.get_backend_name()
if backend == "mysql":
backend = "mariadb"
new_url = copy.copy(url)
+ new_url.query = dict(new_url.query)
new_url.drivername = "%s+%s" % (backend, driver)
+ new_url.query.update(query)
+
try:
new_url.get_dialect()
except exc.NoSuchModuleError:
:name: pg8000
:dbapi: pg8000
:connectstring: postgresql+pg8000://user:password@host:port/dbname[?key=value&key=value...]
- :url: https://pythonhosted.org/pg8000/
+ :url: https://pypi.org/project/pg8000/
-.. note::
-
- The pg8000 dialect is **not tested as part of SQLAlchemy's continuous
- integration** and may have unresolved issues. The recommended PostgreSQL
- dialect is psycopg2.
+.. versionchanged:: 1.4 The pg8000 dialect has been updated for version
+ 1.16.5 and higher, and is again part of SQLAlchemy's continuous integration
+ with full feature support.
.. _pg8000_unicode:
* ``SERIALIZABLE``
* ``AUTOCOMMIT``
-.. versionadded:: 0.9.5 support for AUTOCOMMIT isolation level when using
- pg8000.
-
.. seealso::
:ref:`postgresql_isolation_level`
from .base import _DECIMAL_TYPES
from .base import _FLOAT_TYPES
from .base import _INT_TYPES
+from .base import ENUM
+from .base import INTERVAL
from .base import PGCompiler
from .base import PGDialect
from .base import PGExecutionContext
from .base import PGIdentifierPreparer
from .base import UUID
from .json import JSON
+from .json import JSONB
+from .json import JSONPathType
from ... import exc
from ... import processors
from ... import types as sqltypes
else:
return super(_PGJSON, self).result_processor(dialect, coltype)
+ def get_dbapi_type(self, dbapi):
+ return dbapi.JSON
+
+
+class _PGJSONB(JSONB):
+ def result_processor(self, dialect, coltype):
+ if dialect._dbapi_version > (1, 10, 1):
+ return None # Has native JSON
+ else:
+ return super(_PGJSON, self).result_processor(dialect, coltype)
+
+ def get_dbapi_type(self, dbapi):
+ return dbapi.JSONB
+
+
+class _PGJSONIndexType(sqltypes.JSON.JSONIndexType):
+ def get_dbapi_type(self, dbapi):
+ raise NotImplementedError("should not be here")
+
+
+class _PGJSONIntIndexType(sqltypes.JSON.JSONIntIndexType):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.INTEGER
+
+
+class _PGJSONStrIndexType(sqltypes.JSON.JSONStrIndexType):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.STRING
+
+
+class _PGJSONPathType(JSONPathType):
+ def get_dbapi_type(self, dbapi):
+ return 1009
+
class _PGUUID(UUID):
def bind_processor(self, dialect):
return process
+class _PGEnum(ENUM):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.UNKNOWN
+
+
+class _PGInterval(INTERVAL):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.INTERVAL
+
+ @classmethod
+ def adapt_emulated_to_native(cls, interval, **kw):
+ return _PGInterval(precision=interval.second_precision)
+
+
+class _PGTimeStamp(sqltypes.DateTime):
+ def get_dbapi_type(self, dbapi):
+ if self.timezone:
+ # TIMESTAMPTZOID
+ return 1184
+ else:
+ # TIMESTAMPOID
+ return 1114
+
+
+class _PGTime(sqltypes.Time):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.TIME
+
+
+class _PGInteger(sqltypes.Integer):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.INTEGER
+
+
+class _PGSmallInteger(sqltypes.SmallInteger):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.INTEGER
+
+
+class _PGNullType(sqltypes.NullType):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.NULLTYPE
+
+
+class _PGBigInteger(sqltypes.BigInteger):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.BIGINTEGER
+
+
+class _PGBoolean(sqltypes.Boolean):
+ def get_dbapi_type(self, dbapi):
+ return dbapi.BOOLEAN
+
+
class PGExecutionContext_pg8000(PGExecutionContext):
- pass
+ def pre_exec(self):
+ if not self.compiled:
+ return
+
+ if self.dialect._dbapi_version > (1, 16, 0):
+ self.set_input_sizes()
class PGCompiler_pg8000(PGCompiler):
+ self.process(binary.right, **kw)
)
- def post_process_text(self, text):
- if "%%" in text:
- util.warn(
- "The SQLAlchemy postgresql dialect "
- "now automatically escapes '%' in text() "
- "expressions to '%%'."
- )
- return text.replace("%", "%%")
-
class PGIdentifierPreparer_pg8000(PGIdentifierPreparer):
- def _escape_identifier(self, value):
- value = value.replace(self.escape_quote, self.escape_to_quote)
- return value.replace("%", "%%")
+ def __init__(self, *args, **kwargs):
+ PGIdentifierPreparer.__init__(self, *args, **kwargs)
+ self._double_percents = False
class PGDialect_pg8000(PGDialect):
{
sqltypes.Numeric: _PGNumericNoBind,
sqltypes.Float: _PGNumeric,
- JSON: _PGJSON,
sqltypes.JSON: _PGJSON,
+ sqltypes.Boolean: _PGBoolean,
+ sqltypes.NullType: _PGNullType,
+ JSONB: _PGJSONB,
+ sqltypes.JSON.JSONPathType: _PGJSONPathType,
+ sqltypes.JSON.JSONIndexType: _PGJSONIndexType,
+ sqltypes.JSON.JSONIntIndexType: _PGJSONIntIndexType,
+ sqltypes.JSON.JSONStrIndexType: _PGJSONStrIndexType,
UUID: _PGUUID,
+ sqltypes.Interval: _PGInterval,
+ INTERVAL: _PGInterval,
+ sqltypes.DateTime: _PGTimeStamp,
+ sqltypes.Time: _PGTime,
+ sqltypes.Integer: _PGInteger,
+ sqltypes.SmallInteger: _PGSmallInteger,
+ sqltypes.BigInteger: _PGBigInteger,
+ sqltypes.Enum: _PGEnum,
},
)
fns.append(on_connect)
+ if self._dbapi_version > (1, 16, 0) and self._json_deserializer:
+
+ def on_connect(conn):
+ # json
+ conn.register_in_adapter(114, self._json_deserializer)
+
+ # jsonb
+ conn.register_in_adapter(3802, self._json_deserializer)
+
+ fns.append(on_connect)
+
if len(fns) > 0:
def on_connect(conn):
-import copy
import time
from ... import exc
from ... import text
from ...testing.provision import create_db
from ...testing.provision import drop_db
-from ...testing.provision import generate_driver_url
from ...testing.provision import log
from ...testing.provision import temp_table_keyword_args
-@generate_driver_url.for_db("postgresql")
-def generate_driver_url(url, driver):
- new_url = copy.copy(url)
- new_url.drivername = "postgresql+%s" % driver
- if new_url.get_driver_name() == "asyncpg":
- new_url.query = dict(new_url.query)
- new_url.query["async_fallback"] = "true"
- try:
- new_url.get_dialect()
- except exc.NoSuchModuleError:
- return None
- else:
- return new_url
-
-
@create_db.for_db("postgresql")
def _pg_create_db(cfg, eng, ident):
template_db = cfg.options.postgresql_templatedb
assert success, "Callable did not raise an exception"
-def expect_raises(except_cls):
- return _expect_raises(except_cls, check_context=True)
+def expect_raises(except_cls, check_context=True):
+ return _expect_raises(except_cls, check_context=check_context)
-def expect_raises_message(except_cls, msg):
- return _expect_raises(except_cls, msg=msg, check_context=True)
+def expect_raises_message(except_cls, msg, check_context=True):
+ return _expect_raises(except_cls, msg=msg, check_context=check_context)
class AssertsCompiledSQL(object):
from .. import exc
from ..engine import url as sa_url
from ..util import compat
+from ..util import parse_qsl
log = logging.getLogger(__name__)
--dburi postgresql://db1 \
--dburi postgresql://db2 \
--dburi postgresql://db2 \
- --dbdriver=psycopg2 --dbdriver=asyncpg
+ --dbdriver=psycopg2 --dbdriver=asyncpg?async_fallback=true
Noting that the default postgresql driver is psycopg2. the output
would be::
main_driver = url.get_driver_name()
extra_drivers.discard(main_driver)
- url = generate_driver_url(url, main_driver)
+ url = generate_driver_url(url, main_driver, {})
yield str(url)
for drv in list(extra_drivers):
- new_url = generate_driver_url(url, drv)
+
+ if "?" in drv:
+
+ driver_only, query_str = drv.split("?", 1)
+
+ query = parse_qsl(query_str)
+ else:
+ driver_only = drv
+ query = {}
+
+ new_url = generate_driver_url(url, driver_only, query)
if new_url:
extra_drivers.remove(drv)
+
yield str(new_url)
@register.init
-def generate_driver_url(url, driver):
+def generate_driver_url(url, driver, query):
backend = url.get_backend_name()
new_url = copy.copy(url)
+ new_url.query = dict(new_url.query)
new_url.drivername = "%s+%s" % (backend, driver)
+ new_url.query.update(query)
try:
new_url.get_dialect()
except exc.NoSuchModuleError:
from ... import Text
from ... import Time
from ... import TIMESTAMP
-from ... import type_coerce
from ... import TypeDecorator
from ... import Unicode
from ... import UnicodeText
and_(name == "r6", cast(col["b"], String) == '"some value"'), "r6"
)
- def test_crit_against_int_basic(self):
- name = self.tables.data_table.c.name
- col = self.tables.data_table.c["data"]
-
- self._test_index_criteria(
- and_(name == "r6", cast(col["a"], String) == "5"), "r6"
- )
-
- def _dont_test_crit_against_string_coerce_type(self):
- name = self.tables.data_table.c.name
- col = self.tables.data_table.c["data"]
-
- self._test_index_criteria(
- and_(
- name == "r6",
- cast(col["b"], String) == type_coerce("some value", JSON),
- ),
- "r6",
- test_literal=False,
- )
-
- def _dont_test_crit_against_int_coerce_type(self):
- name = self.tables.data_table.c.name
- col = self.tables.data_table.c["data"]
-
- self._test_index_criteria(
- and_(name == "r6", cast(col["a"], String) == type_coerce(5, JSON)),
- "r6",
- test_literal=False,
- )
-
__all__ = (
"UnicodeVarcharTest",
t = Table("t", m, Column("c", type_, primary_key=True))
if version:
- dialect = postgresql.dialect()
+ dialect = testing.db.dialect.__class__()
dialect._get_server_version_info = mock.Mock(
return_value=version
)
Column("user_name", VARCHAR(20)),
)
- @testing.fails_on(
- "postgresql+pg8000",
- "pg8000 still doesn't allow single paren without params",
- )
def test_no_params_option(self, connection):
stmt = (
"SELECT '%'"
Column("user_name", VARCHAR(20)),
)
- @testing.fails_on(
- "postgresql+pg8000",
- "pg8000 still doesn't allow single paren without params",
- )
def test_no_params_option(self):
stmt = (
"SELECT '%'"
from sqlalchemy.testing import assert_raises_message_context_ok
from sqlalchemy.testing import engines
from sqlalchemy.testing import eq_
+from sqlalchemy.testing import expect_raises
from sqlalchemy.testing import expect_warnings
from sqlalchemy.testing import fixtures
from sqlalchemy.testing import is_
eq_(conn.execute(select(1)).scalar(), 1)
conn.close()
- def exercise_stale_connection():
+ with expect_raises(engine.dialect.dbapi.Error, check_context=False):
curs = stale_connection.cursor()
curs.execute("select 1")
- assert_raises(engine.dialect.dbapi.Error, exercise_stale_connection)
-
def test_pre_ping_db_stays_shutdown(self):
engine = engines.reconnecting_engine(options={"pool_pre_ping": True})
@property
def json_array_indexes(self):
- return self.json_type + fails_if("+pg8000")
+ return self.json_type
@property
def datetime_literals(self):
"Firebird still has FP inaccuracy even "
"with only four decimal places",
),
- (
- "postgresql+pg8000",
- None,
- None,
- "postgresql+pg8000 has FP inaccuracy even with "
- "only four decimal places ",
- ),
- (
- "postgresql+psycopg2cffi",
- None,
- None,
- "postgresql+psycopg2cffi has FP inaccuracy even with "
- "only four decimal places ",
- ),
]
)
@property
def duplicate_key_raises_integrity_error(self):
- return fails_on("postgresql+pg8000")
+ return exclusions.open()
def _has_pg_extension(self, name):
def check(config):
mock; python_version < '3.3'
importlib_metadata; python_version < '3.8'
postgresql: .[postgresql]
- postgresql: .[postgresql_asyncpg]
+ postgresql: .[postgresql_asyncpg]; python_version >= '3'
+ postgresql: .[postgresql_pg8000]; python_version >= '3'
mysql: .[mysql]
mysql: .[pymysql]
oracle: .[oracle]
sqlite_file: SQLITE={env:TOX_SQLITE_FILE:--db sqlite_file}
postgresql: POSTGRESQL={env:TOX_POSTGRESQL:--db postgresql}
- postgresql: EXTRA_PG_DRIVERS={env:EXTRA_PG_DRIVERS:--dbdriver psycopg2 --dbdriver asyncpg}
+ py3{,5,6,7,8,9,10,11}-postgresql: EXTRA_PG_DRIVERS={env:EXTRA_PG_DRIVERS:--dbdriver psycopg2 --dbdriver asyncpg?async_fallback=true --dbdriver pg8000}
mysql: MYSQL={env:TOX_MYSQL:--db mysql}
mysql: EXTRA_MYSQL_DRIVERS={env:EXTRA_MYSQL_DRIVERS:--dbdriver mysqldb --dbdriver pymysql}
+
mssql: MSSQL={env:TOX_MSSQL:--db mssql}
oracle,mssql,sqlite_file: IDENTS=--write-idents db_idents.txt
projects / thirdparty / sqlalchemy / sqlalchemy.git / commitdiff
