Coverage for /home/martinb/.local/share/virtualenvs/camcops/lib/python3.6/site-packages/sqlalchemy/dialects/mssql/adodbapi.py : 45%

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# mssql/adodbapi.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
8"""
9.. dialect:: mssql+adodbapi
10 :name: adodbapi
11 :dbapi: adodbapi
12 :connectstring: mssql+adodbapi://<username>:<password>@<dsnname>
13 :url: http://adodbapi.sourceforge.net/
15.. note::
17 The adodbapi dialect is not implemented in SQLAlchemy versions 0.6 and
18 above at this time.
20"""
21import datetime
22import sys
24from sqlalchemy import types as sqltypes
25from sqlalchemy import util
26from sqlalchemy.dialects.mssql.base import MSDateTime
27from sqlalchemy.dialects.mssql.base import MSDialect
30class MSDateTime_adodbapi(MSDateTime):
31 def result_processor(self, dialect, coltype):
32 def process(value):
33 # adodbapi will return datetimes with empty time
34 # values as datetime.date() objects.
35 # Promote them back to full datetime.datetime()
36 if type(value) is datetime.date:
37 return datetime.datetime(value.year, value.month, value.day)
38 return value
40 return process
43class MSDialect_adodbapi(MSDialect):
44 supports_sane_rowcount = True
45 supports_sane_multi_rowcount = True
46 supports_unicode = sys.maxunicode == 65535
47 supports_unicode_statements = True
48 driver = "adodbapi"
50 @classmethod
51 def import_dbapi(cls):
52 import adodbapi as module
54 return module
56 colspecs = util.update_copy(
57 MSDialect.colspecs, {sqltypes.DateTime: MSDateTime_adodbapi}
58 )
60 def create_connect_args(self, url):
61 def check_quote(token):
62 if ";" in str(token):
63 token = "'%s'" % token
64 return token
66 keys = dict((k, check_quote(v)) for k, v in url.query.items())
68 connectors = ["Provider=SQLOLEDB"]
69 if "port" in keys:
70 connectors.append(
71 "Data Source=%s, %s" % (keys.get("host"), keys.get("port"))
72 )
73 else:
74 connectors.append("Data Source=%s" % keys.get("host"))
75 connectors.append("Initial Catalog=%s" % keys.get("database"))
76 user = keys.get("user")
77 if user:
78 connectors.append("User Id=%s" % user)
79 connectors.append("Password=%s" % keys.get("password", ""))
80 else:
81 connectors.append("Integrated Security=SSPI")
82 return [[";".join(connectors)], {}]
84 def is_disconnect(self, e, connection, cursor):
85 return isinstance(
86 e, self.dbapi.adodbapi.DatabaseError
87 ) and "'connection failure'" in str(e)
90dialect = MSDialect_adodbapi