\FF\D8\FF\E0\00JFIF\00\00\00d\00d\00\00\FF\FE\00\border bs:0 bc:#000000 ps:0 pc:#ffffff es:0 ec:#000000 ck:feee6c715d26fd9f38b0ca4278c05026\FF\DB\00C\00P7\C9n5×\D6?\BDê\9Ds\EBp\9F[`8m\B7)o\B5\E8\E6I\99\FE3]]A2\BA\8Cw\D6E\93\\DEv\C8\009\F2\F1NI?uc\\F5\EA\96k\xN<~buv\EA\C8\D7 \8B\84\CEcxI\BBg\AE\9E=\D6+n\EC\80\C8A\8C\AE\EB\CF\D5\DA\E9"2\A4\B9j5\EB\F3W\B63\96\B30Yu\DA\FC8\ED\DF\E7Ms\FB\F1\8E\B3\FA\EA\E8\E6(\883zs\F2_\8DFk\8Bh \00\8C\DCw\D3R\B5+6X\BA\B2\C4j\AB0\B4\FCMw\C2I\8E\9B\E3\A9~9u\FA\D3l\80\C8%p\EE\FDn2€ \00 $\FEj\C4e\A9\DB\~\95\A7\A5\80EK\BB\8DDsP\00@@AD'k\CF\E8\DB\D2(\80\9AK\D3\85\D6lb\F2\BA\8C*\80\00)\95 59\A3R:\F3\CE"\B6\80\88\00\00i1u4ê\E9\F2á\A6\A2\FACM\93WMb*\E0*\00\00\00\00\00(\A8\80\00\00\80\00\00\00\00\00\00\00\00\00\FF\D9 C/// # connectors/pyodbc.py # Copyright (C) 2005-2021 the SQLAlchemy authors and contributors # # # This module is part of SQLAlchemy and is released under # the MIT License: http://www.opensource.org/licenses/mit-license.php import re from . import Connector from .. import util class PyODBCConnector(Connector): driver = "pyodbc" # this is no longer False for pyodbc in general supports_sane_rowcount_returning = True supports_sane_multi_rowcount = False supports_unicode_statements = True supports_unicode_binds = True supports_native_decimal = True default_paramstyle = "named" # for non-DSN connections, this *may* be used to # hold the desired driver name pyodbc_driver_name = None def __init__(self, supports_unicode_binds=None, **kw): super(PyODBCConnector, self).__init__(**kw) if supports_unicode_binds is not None: self.supports_unicode_binds = supports_unicode_binds @classmethod def dbapi(cls): return __import__("pyodbc") def create_connect_args(self, url): opts = url.translate_connect_args(username="user") opts.update(url.query) keys = opts query = url.query connect_args = {} for param in ("ansi", "unicode_results", "autocommit"): if param in keys: connect_args[param] = util.asbool(keys.pop(param)) if "odbc_connect" in keys: connectors = [util.unquote_plus(keys.pop("odbc_connect"))] else: def check_quote(token): if ";" in str(token): token = "{%s}" % token.replace("}", "}}") return token keys = dict((k, check_quote(v)) for k, v in keys.items()) dsn_connection = "dsn" in keys or ( "host" in keys and "database" not in keys ) if dsn_connection: connectors = [ "dsn=%s" % (keys.pop("host", "") or keys.pop("dsn", "")) ] else: port = "" if "port" in keys and "port" not in query: port = ",%d" % int(keys.pop("port")) connectors = [] driver = keys.pop("driver", self.pyodbc_driver_name) if driver is None and keys: # note if keys is empty, this is a totally blank URL util.warn( "No driver name specified; " "this is expected by PyODBC when using " "DSN-less connections" ) else: connectors.append("DRIVER={%s}" % driver) connectors.extend( [ "Server=%s%s" % (keys.pop("host", ""), port), "Database=%s" % keys.pop("database", ""), ] ) user = keys.pop("user", None) if user: connectors.append("UID=%s" % user) pwd = keys.pop("password", "") if pwd: connectors.append("PWD=%s" % pwd) else: authentication = keys.pop("authentication", None) if authentication: connectors.append("Authentication=%s" % authentication) else: connectors.append("Trusted_Connection=Yes") # if set to 'Yes', the ODBC layer will try to automagically # convert textual data from your database encoding to your # client encoding. This should obviously be set to 'No' if # you query a cp1253 encoded database from a latin1 client... if "odbc_autotranslate" in keys: connectors.append( "AutoTranslate=%s" % keys.pop("odbc_autotranslate") ) connectors.extend(["%s=%s" % (k, v) for k, v in keys.items()]) return [[";".join(connectors)], connect_args] def is_disconnect(self, e, connection, cursor): if isinstance(e, self.dbapi.ProgrammingError): return "The cursor's connection ha