Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/postgresql/pypostgresql.py : 63%

Hot-keys on this page
r m x p toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# postgresql/pypostgresql.py
2# Copyright (C) 2005-2020 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/
14.. note::
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.
21""" # noqa
23from .base import PGDialect
24from .base import PGExecutionContext
25from ... import processors
26from ... import types as sqltypes
27from ... import util
30class PGNumeric(sqltypes.Numeric):
31 def bind_processor(self, dialect):
32 return processors.to_str
34 def result_processor(self, dialect, coltype):
35 if self.asdecimal:
36 return None
37 else:
38 return processors.to_float
41class PGExecutionContext_pypostgresql(PGExecutionContext):
42 pass
45class PGDialect_pypostgresql(PGDialect):
46 driver = "pypostgresql"
48 supports_unicode_statements = True
49 supports_unicode_binds = True
50 description_encoding = None
51 default_paramstyle = "pyformat"
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
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 )
68 @classmethod
69 def dbapi(cls):
70 from postgresql.driver import dbapi20
72 return dbapi20
74 _DBAPI_ERROR_NAMES = [
75 "Error",
76 "InterfaceError",
77 "DatabaseError",
78 "DataError",
79 "OperationalError",
80 "IntegrityError",
81 "InternalError",
82 "ProgrammingError",
83 "NotSupportedError",
84 ]
86 @util.memoized_property
87 def dbapi_exception_translation_map(self):
88 if self.dbapi is None:
89 return {}
91 return dict(
92 (getattr(self.dbapi, name).__name__, name)
93 for name in self._DBAPI_ERROR_NAMES
94 )
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)
105 def is_disconnect(self, e, connection, cursor):
106 return "connection is closed" in str(e)
109dialect = PGDialect_pypostgresql