diff --git a/ibm_db_sa/ibm_db.py b/ibm_db_sa/ibm_db.py index 9607acd..9ddb9ae 100644 --- a/ibm_db_sa/ibm_db.py +++ b/ibm_db_sa/ibm_db.py @@ -18,33 +18,34 @@ # +--------------------------------------------------------------------------+ import re +import warnings +from collections.abc import MutableMapping, Sequence +from typing import Any, cast + from sqlalchemy import __version__ as SA_VERSION_STR -from .logger import init_ibmdbsa_logging, logger, log_entry_exit +from sqlalchemy import types as sa_types +from sqlalchemy import util +from sqlalchemy.engine import result as _result +from sqlalchemy.engine.url import URL +from sqlalchemy.exc import ArgumentError + +from .base import DB2Dialect, DB2ExecutionContext +from .logger import init_ibmdbsa_logging, log_entry_exit, logger + m = re.match(r"^\s*(\d+)\.(\d+)", SA_VERSION_STR) SA_VERSION_MM = (int(m.group(1)), int(m.group(2))) if m else (0, 0) -from .base import DB2ExecutionContext, DB2Dialect - if SA_VERSION_MM < (2, 0): - from sqlalchemy import processors, types as sa_types, util + from sqlalchemy import processors else: - from sqlalchemy import types as sa_types, util from sqlalchemy.engine import processors -from sqlalchemy.exc import ArgumentError - SQL_TXN_READ_UNCOMMITTED = 1 SQL_TXN_READ_COMMITTED = 2 SQL_TXN_REPEATABLE_READ = 4 SQL_TXN_SERIALIZABLE = 8 SQL_ATTR_TXN_ISOLATION = 108 -if SA_VERSION_MM < (0, 8): - from sqlalchemy.engine import base -else: - from sqlalchemy.engine import result as _result - - class _IBM_Numeric_ibm_db(sa_types.Numeric): @log_entry_exit def result_processor(self, dialect, coltype): @@ -234,65 +235,63 @@ def reset_isolation_level(self, connection): logger.debug("Resetting isolation level to default (CS)") self.set_isolation_level(connection, 'CS') - def create_connect_args(self, url): - url, ibmdbsa_log_value = init_ibmdbsa_logging(url) - # DSN support through CLI configuration (../cfg/db2cli.ini), - # while 2 connection attributes are mandatory: database alias - # and UID (in support to current schema), all the other - # connection attributes (protocol, hostname, servicename) are - # provided through db2cli.ini database catalog entry. Example - # 1: ibm_db_sa:///?UID=db2inst1 or Example 2: - # ibm_db_sa:///?DSN=;UID=db2inst1 + def create_connect_args( + self, url: URL + ) -> tuple[Sequence[str], MutableMapping[str, Any]]: + url, _ = cast(tuple[URL, None], init_ibmdbsa_logging(url)) logger.info("entry create_connect_args()") + + if url.database is None: + raise ValueError(f"the engine URL `{url!r}` does not specify a database") + if not url.host: logger.debug("Using DSN based connection") - dsn = url.database - uid = url.username - pwd = url.password - logger.debug("DSN connection parameters -> database=%s user=%s", dsn, uid) + username = url.username or "" + logger.debug( + "DSN connection parameters -> database=%s user=%s", + url.database, + username, + ) logger.info("exit create_connect_args()") - return (dsn, uid, pwd, '', ''), {} + return (url.database, username, url.password or "", "", ""), {} + + logger.debug("Using full connection URL for remote DB2 server") + params: dict[str, Any] = url.translate_connect_args( + database="DATABASE", + host="HOSTNAME", + password="PWD", + port="PORT", + username="UID", + ) + for k, v in params.items(): + if k == "PWD": + continue + logger.debug(f"{k}: %s", v) + params["PROTOCOL"] = "TCPIP" + for k, v in url.query.items(): + logger.debug("Applying connection option: %s=%s", k, v) + # This doesn't handle the tuple of strings scenario, but neither does the original code. + params |= url.query + + if url.password is None: + safe_params = params else: - # Full URL string support for connection to remote data servers - logger.debug("Using full connection URL for remote DB2 server") - dsn_param = ['DATABASE=%s' % url.database, - 'HOSTNAME=%s' % url.host, - 'PROTOCOL=TCPIP'] - logger.debug("Host: %s", url.host) - logger.debug("Database: %s", url.database) - if url.port: - logger.debug("Port: %s", url.port) - dsn_param.append('PORT=%s' % url.port) - if url.username: - logger.debug("User: %s", url.username) - dsn_param.append('UID=%s' % url.username) - if url.password: - # if password contains ';' truncate at first ';' (existing logic) - if ';' in url.password: - logger.debug("Password contains ';', truncating") - url = url._replace(password=(url.password).partition(";")[0]) - dsn_param.append('PWD=%s' % url.password) - # check for connection arguments - connection_keys = ['Security', 'SSLClientKeystoredb', 'SSLClientKeystash', 'SSLServerCertificate', - 'CurrentSchema'] - # rebuild query_keys in case url changed - query_keys = list(url.query.keys()) if url.query else [] - for key in connection_keys: - for query_key in query_keys: - if query_key.lower() == key.lower(): - logger.debug("Applying connection option: %s=%s", key, url.query[query_key]) - dsn_param.append( - '%(connection_key)s=%(value)s' % {'connection_key': key, 'value': url.query[query_key]}) - url = url.difference_update_query([query_key]) - break - dsn = ';'.join(dsn_param) - dsn += ';' - safe_dsn = dsn - if 'PWD=' in safe_dsn: - safe_dsn = re.sub(r'PWD=[^;]*', 'PWD=****', safe_dsn) - logger.debug("Constructed DB2 DSN: %s", safe_dsn) - logger.info("exit create_connect_args()") - return (dsn, url.username, '', '', ''), {} + safe_params = params | {"PWD": "****"} + truncated_password, semicolon, _ = cast(str, url.password).partition(";") + if semicolon: + logger.debug("Password contains ';', truncating") + warnings.warn( + "The provided password contained ';' and has been truncated. " + "This will become an exception in a future release.", + DeprecationWarning, + ) + params["PWD"] = truncated_password + + safe_dsn = "".join(f"{k}={v};" for k, v in safe_params.items()) + logger.debug("Constructed DB2 DSN: %s", safe_dsn) + dsn = "".join(f"{k}={v};" for k, v in params.items()) + logger.info("exit create_connect_args()") + return (dsn, "", "", "", ""), {} # Retrieves current schema for the specified connection object @log_entry_exit diff --git a/setup.py b/setup.py index d44a02c..7126bcb 100644 --- a/setup.py +++ b/setup.py @@ -10,10 +10,11 @@ v.close() readme = os.path.join(os.path.dirname(__file__), 'README.md') +sqlalchemy_requirement = "sqlalchemy>=1.3.5" if 'USE_PYODBC' in os.environ and os.environ['USE_PYODBC'] == '1': - require = ['sqlalchemy>=0.7.3'] + require = [sqlalchemy_requirement] else: - require = ['sqlalchemy>=0.7.3','ibm_db>=2.0.0'] + require = [sqlalchemy_requirement,'ibm_db>=2.0.0'] setup( @@ -37,6 +38,7 @@ long_description=open(readme).read(), platforms='All', install_requires= require, + python_requires=">=3.9", packages=['ibm_db_sa'], entry_points={ 'sqlalchemy.dialects': [