1# postgresql/pypostgresql.py
2# Copyright (C) 2005-2021 the SQLAlchemy authors and contributors
3# <see AUTHORS file>
4#
5# This module is part of SQLAlchemy and is released under
6# the MIT License: http://www.opensource.org/licenses/mit-license.php
7"""
8.. dialect:: postgresql+pypostgresql
9    :name: py-postgresql
10    :dbapi: pypostgresql
11    :connectstring: postgresql+pypostgresql://user:password@host:port/dbname[?key=value&key=value...]
12    :url: http://python.projects.pgfoundry.org/
13
14.. note::
15
16    The pypostgresql dialect is **not tested as part of SQLAlchemy's continuous
17    integration** and may have unresolved issues.  The recommended PostgreSQL
18    driver is psycopg2.
19
20
21"""  # noqa
22
23from .base import PGDialect
24from .base import PGExecutionContext
25from ... import processors
26from ... import types as sqltypes
27from ... import util
28
29
30class PGNumeric(sqltypes.Numeric):
31    def bind_processor(self, dialect):
32        return processors.to_str
33
34    def result_processor(self, dialect, coltype):
35        if self.asdecimal:
36            return None
37        else:
38            return processors.to_float
39
40
41class PGExecutionContext_pypostgresql(PGExecutionContext):
42    pass
43
44
45class PGDialect_pypostgresql(PGDialect):
46    driver = "pypostgresql"
47
48    supports_unicode_statements = True
49    supports_unicode_binds = True
50    description_encoding = None
51    default_paramstyle = "pyformat"
52
53    # requires trunk version to support sane rowcounts
54    # TODO: use dbapi version information to set this flag appropriately
55    supports_sane_rowcount = True
56    supports_sane_multi_rowcount = False
57
58    execution_ctx_cls = PGExecutionContext_pypostgresql
59    colspecs = util.update_copy(
60        PGDialect.colspecs,
61        {
62            sqltypes.Numeric: PGNumeric,
63            # prevents PGNumeric from being used
64            sqltypes.Float: sqltypes.Float,
65        },
66    )
67
68    @classmethod
69    def dbapi(cls):
70        from postgresql.driver import dbapi20
71
72        return dbapi20
73
74    _DBAPI_ERROR_NAMES = [
75        "Error",
76        "InterfaceError",
77        "DatabaseError",
78        "DataError",
79        "OperationalError",
80        "IntegrityError",
81        "InternalError",
82        "ProgrammingError",
83        "NotSupportedError",
84    ]
85
86    @util.memoized_property
87    def dbapi_exception_translation_map(self):
88        if self.dbapi is None:
89            return {}
90
91        return dict(
92            (getattr(self.dbapi, name).__name__, name)
93            for name in self._DBAPI_ERROR_NAMES
94        )
95
96    def create_connect_args(self, url):
97        opts = url.translate_connect_args(username="user")
98        if "port" in opts:
99            opts["port"] = int(opts["port"])
100        else:
101            opts["port"] = 5432
102        opts.update(url.query)
103        return ([], opts)
104
105    def is_disconnect(self, e, connection, cursor):
106        return "connection is closed" in str(e)
107
108
109dialect = PGDialect_pypostgresql
110