1# mssql/adodbapi.py 2# Copyright (C) 2005-2019 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""" 9.. dialect:: mssql+adodbapi 10 :name: adodbapi 11 :dbapi: adodbapi 12 :connectstring: mssql+adodbapi://<username>:<password>@<dsnname> 13 :url: http://adodbapi.sourceforge.net/ 14 15.. note:: 16 17 The adodbapi dialect is not implemented SQLAlchemy versions 0.6 and 18 above at this time. 19 20""" 21import datetime 22import sys 23 24from sqlalchemy import types as sqltypes 25from sqlalchemy import util 26from sqlalchemy.dialects.mssql.base import MSDateTime 27from sqlalchemy.dialects.mssql.base import MSDialect 28 29 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 39 40 return process 41 42 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" 49 50 @classmethod 51 def import_dbapi(cls): 52 import adodbapi as module 53 54 return module 55 56 colspecs = util.update_copy( 57 MSDialect.colspecs, {sqltypes.DateTime: MSDateTime_adodbapi} 58 ) 59 60 def create_connect_args(self, url): 61 def check_quote(token): 62 if ";" in str(token): 63 token = "'%s'" % token 64 return token 65 66 keys = dict((k, check_quote(v)) for k, v in url.query.items()) 67 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)], {}] 83 84 def is_disconnect(self, e, connection, cursor): 85 return isinstance( 86 e, self.dbapi.adodbapi.DatabaseError 87 ) and "'connection failure'" in str(e) 88 89 90dialect = MSDialect_adodbapi 91