Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

adodbapi: Simplify dict iterations from Python 2 to 3 migration #2332

Open
wants to merge 12 commits into
base: main
Choose a base branch
from
2 changes: 1 addition & 1 deletion CHANGES.txt
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@ Coming in build 309, as yet unreleased
Import `DispatcherWin32trace` and `DispatcherTrace` from `win32com.server.dispatcher` instead.
* Fixed `win32timezone.TimeZoneInfo` initialization from a `[DYNAMIC_]TIME_ZONE_INFORMATION` (#2339, @Avasam)
* Added runtime deprecation warning of `win2kras`, use `win32ras` instead (#2356, @Avasam)
* Improved handling of dict iterations and fallbacks (removes Python 2 support code, small general speed improvement) (#2332, #2330, @Avasam)
* Improved handling of dict iterations and fallbacks (removes Python 2 support code, small general speed improvement) (#2330, #2331, #2332, @Avasam)
* Fixed "Open GL Demo" (`Pythonwin/pywin/Demos/openGLDemo.py`) and restore "Font" demo in `Pythonwin/pywin/Demos/guidemo.py` (#2345, @Avasam)
* Fixed accidentally trying to raise an undefined name instead of an `Exception` in `Pythonwin/pywin/debugger/debugger.py` (#2326, @Avasam)

Expand Down
9 changes: 5 additions & 4 deletions adodbapi/adodbapi.py
Original file line number Diff line number Diff line change
Expand Up @@ -234,7 +234,9 @@ def __init__(self): # now define the instance attributes
self.paramstyle = api.paramstyle
self.supportsTransactions = False
self.connection_string = ""
self.cursors = weakref.WeakValueDictionary()
self.cursors: weakref.WeakValueDictionary[int, Cursor] = (
weakref.WeakValueDictionary()
)
self.dbms_name = ""
self.dbms_version = ""
self.errorhandler = None # use the standard error handler for this instance
Expand Down Expand Up @@ -326,9 +328,8 @@ def close(self):
an Error (or subclass) exception will be raised if any operation is attempted with the connection.
The same applies to all cursor objects trying to use the connection.
"""
for crsr in list(self.cursors.values())[
:
]: # copy the list, then close each one
# copy the list of cursors to avoid size changing during iteration, then close each one
for crsr in list(self.cursors.values()):
crsr.close(dont_tell_me=True) # close without back-link clearing
self.messages = []
try:
Expand Down
2 changes: 1 addition & 1 deletion adodbapi/apibase.py
Original file line number Diff line number Diff line change
Expand Up @@ -564,7 +564,7 @@ def __next__(self):
yield self._getValue(n)

def __repr__(self): # create a human readable representation
taglist = sorted(list(self.rows.columnNames.items()), key=lambda x: x[1])
taglist = sorted(self.rows.columnNames.values())
s = "<SQLrow={"
for name, i in taglist:
s += f"{name}:{self._getValue(i)!r}, "
Expand Down
276 changes: 141 additions & 135 deletions adodbapi/process_connect_string.py
Original file line number Diff line number Diff line change
@@ -1,135 +1,141 @@
"""a clumsy attempt at a macro language to let the programmer execute code on the server (ex: determine 64bit)"""

from . import is64bit


def macro_call(macro_name, args, kwargs):
"""allow the programmer to perform limited processing on the server by passing macro names and args

:new_key - the key name the macro will create
:args[0] - macro name
:args[1:] - any arguments
:code - the value of the keyword item
:kwargs - the connection keyword dictionary. ??key has been removed
--> the value to put in for kwargs['name'] = value
"""
if isinstance(args, (str, str)):
args = [
args
] # the user forgot to pass a sequence, so make a string into args[0]
new_key = args[0]
try:
if macro_name == "is64bit":
if is64bit.Python(): # if on 64 bit Python
return new_key, args[1] # return first argument
else:
try:
return new_key, args[2] # else return second argument (if defined)
except IndexError:
return new_key, "" # else return blank

elif (
macro_name == "getuser"
): # get the name of the user the server is logged in under
if not new_key in kwargs:
import getpass

return new_key, getpass.getuser()

elif macro_name == "getnode": # get the name of the computer running the server
import platform

try:
return new_key, args[1] % platform.node()
except IndexError:
return new_key, platform.node()

elif macro_name == "getenv": # expand the server's environment variable args[1]
try:
dflt = args[2] # if not found, default from args[2]
except IndexError: # or blank
dflt = ""
return new_key, os.environ.get(args[1], dflt)

elif macro_name == "auto_security":
if (
not "user" in kwargs or not kwargs["user"]
): # missing, blank, or Null username
return new_key, "Integrated Security=SSPI"
return new_key, "User ID=%(user)s; Password=%(password)s" % kwargs

elif (
macro_name == "find_temp_test_path"
): # helper function for testing ado operation -- undocumented
import os
import tempfile

return new_key, os.path.join(
tempfile.gettempdir(), "adodbapi_test", args[1]
)

raise ValueError(f"Unknown connect string macro={macro_name}")
except:
raise ValueError(f"Error in macro processing {macro_name} {args!r}")


def process(
args, kwargs, expand_macros=False
): # --> connection string with keyword arguments processed.
"""attempts to inject arguments into a connection string using Python "%" operator for strings

co: adodbapi connection object
args: positional parameters from the .connect() call
kvargs: keyword arguments from the .connect() call
"""
try:
dsn = args[0]
except IndexError:
dsn = None
# as a convenience the first argument may be django settings
if isinstance(dsn, dict):
kwargs.update(dsn)
# the connection string is passed to the connection as part of the keyword dictionary
elif dsn:
kwargs["connection_string"] = dsn
try:
a1 = args[1]
except IndexError:
a1 = None
# historically, the second positional argument might be a timeout value
if isinstance(a1, int):
kwargs["timeout"] = a1
# if the second positional argument is a string, then it is user
elif isinstance(a1, str):
kwargs["user"] = a1
# if the second positional argument is a dictionary, use it as keyword arguments, too
elif isinstance(a1, dict):
kwargs.update(a1)
try:
kwargs["password"] = args[2] # the third positional argument is password
kwargs["host"] = args[3] # the fourth positional argument is host name
kwargs["database"] = args[4] # the fifth positional argument is database name
except IndexError:
pass

# make sure connection string is defined somehow
if not "connection_string" in kwargs:
try: # perhaps 'dsn' was defined
kwargs["connection_string"] = kwargs["dsn"]
except KeyError:
try: # as a last effort, use the "host" keyword
kwargs["connection_string"] = kwargs["host"]
except KeyError:
raise TypeError("Must define 'connection_string' for ado connections")
if expand_macros:
for kwarg in list(kwargs.keys()):
if kwarg.startswith("macro_"): # If a key defines a macro
macro_name = kwarg[6:] # name without the "macro_"
macro_code = kwargs.pop(
kwarg
) # we remove the macro_key and get the code to execute
new_key, rslt = macro_call(
macro_name, macro_code, kwargs
) # run the code in the local context
kwargs[new_key] = rslt # put the result back in the keywords dict
return kwargs
"""a clumsy attempt at a macro language to let the programmer execute code on the server (ex: determine 64bit)"""

from __future__ import annotations

from collections.abc import MutableMapping
from typing import Any

from . import is64bit


def macro_call(macro_name, args, kwargs):
"""allow the programmer to perform limited processing on the server by passing macro names and args

:new_key - the key name the macro will create
:args[0] - macro name
:args[1:] - any arguments
:code - the value of the keyword item
:kwargs - the connection keyword dictionary. ??key has been removed
--> the value to put in for kwargs['name'] = value
"""
if isinstance(args, (str, str)):
args = [
args
] # the user forgot to pass a sequence, so make a string into args[0]
new_key = args[0]
try:
if macro_name == "is64bit":
if is64bit.Python(): # if on 64 bit Python
return new_key, args[1] # return first argument
else:
try:
return new_key, args[2] # else return second argument (if defined)
except IndexError:
return new_key, "" # else return blank

elif (
macro_name == "getuser"
): # get the name of the user the server is logged in under
if not new_key in kwargs:
import getpass

return new_key, getpass.getuser()

elif macro_name == "getnode": # get the name of the computer running the server
import platform

try:
return new_key, args[1] % platform.node()
except IndexError:
return new_key, platform.node()

elif macro_name == "getenv": # expand the server's environment variable args[1]
try:
dflt = args[2] # if not found, default from args[2]
except IndexError: # or blank
dflt = ""
return new_key, os.environ.get(args[1], dflt)

elif macro_name == "auto_security":
if (
not "user" in kwargs or not kwargs["user"]
): # missing, blank, or Null username
return new_key, "Integrated Security=SSPI"
return new_key, "User ID=%(user)s; Password=%(password)s" % kwargs

elif (
macro_name == "find_temp_test_path"
): # helper function for testing ado operation -- undocumented
import os
import tempfile

return new_key, os.path.join(
tempfile.gettempdir(), "adodbapi_test", args[1]
)

raise ValueError(f"Unknown connect string macro={macro_name}")
except:
raise ValueError(f"Error in macro processing {macro_name} {args!r}")


def process(
args, kwargs: MutableMapping[str, Any], expand_macros=False
): # --> connection string with keyword arguments processed.
"""attempts to inject arguments into a connection string using Python "%" operator for strings

co: adodbapi connection object
args: positional parameters from the .connect() call
kvargs: keyword arguments from the .connect() call
"""
try:
dsn = args[0]
except IndexError:
dsn = None
# as a convenience the first argument may be django settings
if isinstance(dsn, dict):
kwargs.update(dsn)
# the connection string is passed to the connection as part of the keyword dictionary
elif dsn:
kwargs["connection_string"] = dsn
try:
a1 = args[1]
except IndexError:
a1 = None
# historically, the second positional argument might be a timeout value
if isinstance(a1, int):
kwargs["timeout"] = a1
# if the second positional argument is a string, then it is user
elif isinstance(a1, str):
kwargs["user"] = a1
# if the second positional argument is a dictionary, use it as keyword arguments, too
elif isinstance(a1, dict):
kwargs.update(a1)
try:
kwargs["password"] = args[2] # the third positional argument is password
kwargs["host"] = args[3] # the fourth positional argument is host name
kwargs["database"] = args[4] # the fifth positional argument is database name
except IndexError:
pass

# make sure connection string is defined somehow
if not "connection_string" in kwargs:
try: # perhaps 'dsn' was defined
kwargs["connection_string"] = kwargs["dsn"]
except KeyError:
try: # as a last effort, use the "host" keyword
kwargs["connection_string"] = kwargs["host"]
except KeyError:
raise TypeError("Must define 'connection_string' for ado connections")
if expand_macros:
# copy the list to avoid size changing during iteration
for kwarg in list(kwargs):
if kwarg.startswith("macro_"): # If a key defines a macro
macro_name = kwarg[6:] # name without the "macro_"
macro_code = kwargs.pop(
kwarg
) # we remove the macro_key and get the code to execute
new_key, rslt = macro_call(
macro_name, macro_code, kwargs
) # run the code in the local context
kwargs[new_key] = rslt # put the result back in the keywords dict
return kwargs
Loading