PGDialect_psycopg2 is a class within the sqlalchemy.dialects.postgresql.psycopg2 module of the SQLAlchemy project.
Amazon Redshift SQLAlchemy Dialect is a SQLAlchemy Dialect that can communicate with the AWS Redshift data store. The SQL is essentially PostgreSQL and requires psycopg2 to properly operate. This project and its code are open sourced under the MIT license.
Amazon Redshift SQLAlchemy Dialect / sqlalchemy_redshift / dialect.py
# dialect.py
import re
from collections import defaultdict, namedtuple
from packaging.version import Version
import pkg_resources
import sqlalchemy as sa
from sqlalchemy import inspect
from sqlalchemy.dialects.postgresql.base import (
PGCompiler, PGDDLCompiler, PGIdentifierPreparer, PGTypeCompiler
)
from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
from sqlalchemy.engine import reflection
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.sql.expression import (
BinaryExpression, BooleanClauseList, Delete
)
from sqlalchemy.types import (
VARCHAR, NullType, SMALLINT, INTEGER, BIGINT,
DECIMAL, REAL, BOOLEAN, CHAR, DATE, TIMESTAMP)
from sqlalchemy.dialects.postgresql import DOUBLE_PRECISION
from .commands import (
CopyCommand, UnloadFromSelect, Format, Compression, Encoding,
CreateLibraryCommand, AlterTableAppendCommand, RefreshMaterializedView
)
from .ddl import (
CreateMaterializedView, DropMaterializedView, get_table_attributes
)
sa_version = Version(sa.__version__)
try:
import alembic
except ImportError:
pass
## ... source file continues with no further PGDialect_psycopg2 examples...
databases (project homepage and PyPI page provides asyncio support with an SQLALchemy Core interface for common relational databases such as MySQL, PostgreSQL and SQLite. This is handy for integrating with asynchronous I/O web frameworks like Sanic. The project is open sourced under the BSD 3-Clause "New" or "Revised" License.
databases / databases / backends / aiopg.py
# aiopg.py
import getpass
import json
import logging
import typing
import uuid
import aiopg
from aiopg.sa.engine import APGCompiler_psycopg2
from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
from sqlalchemy.engine.interfaces import Dialect, ExecutionContext
from sqlalchemy.engine.result import ResultMetaData, RowProxy
from sqlalchemy.sql import ClauseElement
from sqlalchemy.types import TypeEngine
from databases.core import DatabaseURL
from databases.interfaces import ConnectionBackend, DatabaseBackend, TransactionBackend
logger = logging.getLogger("databases")
class AiopgBackend(DatabaseBackend):
def __init__(
self, database_url: typing.Union[DatabaseURL, str], **options: typing.Any
) -> None:
self._database_url = DatabaseURL(database_url)
self._options = options
self._dialect = self._get_dialect()
self._pool = None
def _get_dialect(self) -> Dialect:
dialect = PGDialect_psycopg2(
json_serializer=json.dumps, json_deserializer=lambda x: x
)
dialect.statement_compiler = APGCompiler_psycopg2
dialect.implicit_returning = True
dialect.supports_native_enum = True
dialect.supports_smallserial = True # 9.2+
dialect._backslash_escapes = False
dialect.supports_sane_multi_rowcount = True # psycopg 2.0.9+
dialect._has_native_hstore = True
dialect.supports_native_decimal = True
return dialect
def _get_connection_kwargs(self) -> dict:
url_options = self._database_url.options
kwargs = {}
min_size = url_options.get("min_size")
max_size = url_options.get("max_size")
ssl = url_options.get("ssl")
if min_size is not None:
kwargs["minsize"] = int(min_size)
if max_size is not None:
## ... source file continues with no further PGDialect_psycopg2 examples...
sqlalchemy-utils (project documentation and PyPI package information) is a code library with various helper functions and new data types that make it easier to use SQLAlchemy when building projects that involve more specific storage requirements such as currency. The wide array of data types includes ranged values and aggregated attributes.
sqlalchemy-utils / sqlalchemy_utils / types / pg_composite.py
# pg_composite.py
from collections import namedtuple
import six
import sqlalchemy as sa
from sqlalchemy.dialects.postgresql import ARRAY
from sqlalchemy.dialects.postgresql.psycopg2 import PGDialect_psycopg2
from sqlalchemy.ext.compiler import compiles
from sqlalchemy.schema import _CreateDropBase
from sqlalchemy.sql.expression import FunctionElement
from sqlalchemy.types import (
SchemaType,
to_instance,
TypeDecorator,
UserDefinedType
)
from .. import ImproperlyConfigured
psycopg2 = None
CompositeCaster = None
adapt = None
AsIs = None
register_adapter = None
try:
import psycopg2
from psycopg2.extras import CompositeCaster
from psycopg2.extensions import adapt, AsIs, register_adapter
except ImportError:
pass
## ... source file abbreviated to get to PGDialect_psycopg2 examples ...
bind.execute(CreateCompositeType(self))
def drop(self, bind=None, checkfirst=True):
if (
checkfirst and
bind.dialect.has_type(bind, self.name, schema=self.schema)
):
bind.execute(DropCompositeType(self))
def register_psycopg2_composite(dbapi_connection, composite):
psycopg2.extras.register_composite(
composite.name,
dbapi_connection,
globally=True,
factory=composite.caster
)
def adapt_composite(value):
adapted = [
adapt(
getattr(value, column.name)
if not isinstance(column.type, TypeDecorator)
else column.type.process_bind_param(
getattr(value, column.name),
PGDialect_psycopg2()
)
)
for column in
composite.columns
]
for value in adapted:
if hasattr(value, 'prepare'):
value.prepare(dbapi_connection)
values = [
value.getquoted().decode(dbapi_connection.encoding)
if six.PY3
else value.getquoted()
for value in adapted
]
return AsIs("(%s)::%s" % (', '.join(values), composite.name))
register_adapter(composite.type_cls, adapt_composite)
def before_create(target, connection, **kw):
for name, composite in registered_composites.items():
composite.create(connection, checkfirst=True)
register_psycopg2_composite(
connection.connection.connection,
## ... source file continues with no further PGDialect_psycopg2 examples...