')
- for factory in factories:
- wr(f'Flushing cache of {factory.name()}...
')
- factory.flushCache()
- wr('
The caches of all factories' - ' have been flushed.
') - wr('Click here to view the Servlet cache:' - ' Servlet Cache
') + case "Clear cache": + from URLParser import ServletFactoryManager + factories = [f for f in ServletFactoryManager._factories + if f._classCache] + wr('')
+ for factory in factories:
+ wr(f'Flushing cache of {factory.name()}...
')
+ factory.flushCache()
+ wr('
The caches of all factories' + ' have been flushed.
') + wr('Click here to view the Servlet cache:' + ' Servlet Cache
') - elif action == "Reload": - wr('Reloading selected modules. Any existing classes' - ' will continue to use the old module definitions,' - ' as will any functions/variables imported using "from".' - ' Use "Clear Cache" to clean out any servlets' - ' in this condition.
') - reloadNames = req.field("reloads", None) - if not isinstance(reloadNames, list): - reloadNames = [reloadNames] - wr('
')
- for n in reloadNames:
- m = sys.modules.get(n)
- if m:
- wr(f"Reloading {self.htmlEncode(str(m))}...
")
- try:
- reload(m)
- except Exception as e:
- wr('Could not reload, '
- f'error was "{e}".
')
- wr('
The selected modules' - ' have been reloaded.
') + case "Reload": + wr('Reloading selected modules. Any existing classes' + ' will continue to use the old module definitions,' + ' as will any functions/variables imported using "from".' + ' Use "Clear Cache" to clean out any servlets' + ' in this condition.
') + reloadNames = req.field("reloads", None) + if not isinstance(reloadNames, list): + reloadNames = [reloadNames] + wr('
')
+ for n in reloadNames:
+ m = sys.modules.get(n)
+ if m:
+ wr(f"Reloading {self.htmlEncode(str(m))}...
")
+ try:
+ reload(m)
+ except Exception as e:
+ wr('Could not reload, '
+ f'error was "{e}".
')
+ wr('
The selected modules' + ' have been reloaded.
') - else: - wr(f'Cannot perform "{action}".
') + case _: + wr(f'Cannot perform "{action}".
') diff --git a/webware/Admin/Errors.py b/webware/Admin/Errors.py index 621d051..656c20e 100644 --- a/webware/Admin/Errors.py +++ b/webware/Admin/Errors.py @@ -18,9 +18,7 @@ def cellContents(self, _rowIndex, colIndex, value): if self._headings[colIndex] in ('pathname', 'error report filename'): path = self.application().serverSidePath() if value.startswith(path): - value = value[len(path):] - if value.startswith(sep): - value = value[len(sep):] + value = value.removeprefix(path).removeprefix(sep) link = f'View?filename={urlEncode(value)}' value = value.replace(sep, sep + 'Stopped pushing contents.
') else: diff --git a/webware/Examples/YattagDemo.py b/webware/Examples/YattagDemo.py index a950fad..4bb6904 100644 --- a/webware/Examples/YattagDemo.py +++ b/webware/Examples/YattagDemo.py @@ -63,9 +63,8 @@ def demoForm(self): with tag('p', klass='success'): text('Congratulations! You have sent the following message:') with tag('div', klass='output'): - with tag('p'): - with tag('strong'): - text(subject) + with tag('p'), tag('strong'): + text(subject) with tag('p'): text(message) with tag('a', href='YattagDemo'): diff --git a/webware/ExceptionHandler.py b/webware/ExceptionHandler.py index 0cd2ae0..af06c81 100644 --- a/webware/ExceptionHandler.py +++ b/webware/ExceptionHandler.py @@ -506,7 +506,7 @@ def emailException(self, htmlErrMsg): headers['Subject'] = '{} {}: {}'.format( headers.get('Subject', '[Webware Error]'), *sys.exc_info()[:2]) for header, value in headers.items(): - if isinstance(value, (list, tuple)): + if isinstance(value, list | tuple): value = ','.join(value) message.add_header(header, value) diff --git a/webware/HTTPRequest.py b/webware/HTTPRequest.py index f9d50cc..a4d9af8 100644 --- a/webware/HTTPRequest.py +++ b/webware/HTTPRequest.py @@ -351,8 +351,7 @@ def serverSidePath(self, path=None): server side directory to form a path relative to the object. """ if path: - if path.startswith('/'): - path = path[1:] + path = path.removeprefix('/') return os.path.normpath(os.path.join( os.path.dirname(self._serverSidePath), path)) return self._serverSidePath @@ -367,8 +366,7 @@ def serverSideContextPath(self, path=None): if the request is in a subdirectory of the main context directory. """ if path: - if path.startswith('/'): - path = path[1:] + path = path.removeprefix('/') return os.path.normpath(os.path.join( self._serverSideContextPath, path)) return self._serverSideContextPath @@ -385,15 +383,11 @@ def servletURI(self): """Return servlet URI without any query strings or extra path info.""" p = self._pathInfo if not self._extraURLPath: - if p.endswith('/'): - p = p[:-1] - return p + return p.removesuffix('/') i = p.rfind(self._extraURLPath) if i >= 0: p = p[:i] - if p.endswith('/'): - p = p[:-1] - return p + return p.removesuffix('/') def uriWebwareRoot(self): """Return relative URL path of the Webware root location.""" @@ -494,11 +488,9 @@ def siteRoot(self): of the servlet that you have forwarded to. """ url = self.originalURLPath() - if url.startswith('/'): - url = url[1:] + url = url.removeprefix('/') contextName = self.contextName() + '/' - if url.startswith(contextName): - url = url[len(contextName):] + url = url.removeprefix(contextName) numStepsBack = url.count('/') return '../' * numStepsBack @@ -512,11 +504,9 @@ def siteRootFromCurrentServlet(self): relative to the _current_ servlet, not the _original_ servlet. """ url = self.urlPath() - if url.startswith('/'): - url = url[1:] + url = url.removeprefix('/') contextName = self.contextName() + '/' - if url.startswith(contextName): - url = url[len(contextName):] + url = url.removeprefix(contextName) numStepsBackward = url.count('/') return '../' * numStepsBackward @@ -529,8 +519,7 @@ def servletPathFromSiteRoot(self): servlet in a database, for example. """ urlPath = self.urlPath() - if urlPath.startswith('/'): - urlPath = urlPath[1:] + urlPath = urlPath.removeprefix('/') parts = urlPath.split('/') newParts = [] for part in parts: diff --git a/webware/HTTPResponse.py b/webware/HTTPResponse.py index bed574a..9dfcd0b 100644 --- a/webware/HTTPResponse.py +++ b/webware/HTTPResponse.py @@ -123,24 +123,25 @@ def setCookie( cookie = Cookie(name, value) t = expires if isinstance(t, str): - if t == 'ONCLOSE': - t = None - elif t == 'NOW': - cookie.delete() - return - elif t == 'NEVER': - t = gmtime() - t = (t[0] + 10,) + t[1:] - elif t.startswith('+'): - t = time() + timeDecode(t[1:]) + match t: + case 'ONCLOSE': + t = None + case 'NOW': + cookie.delete() + return + case 'NEVER': + t = gmtime() + t = (t[0] + 10,) + t[1:] + case _ if t.startswith('+'): + t = time() + timeDecode(t[1:]) if t: - if isinstance(t, (int, float)): + if isinstance(t, int | float): t = gmtime(t) - if isinstance(t, (tuple, struct_time)): + elif isinstance(t, tuple | struct_time): t = strftime("%a, %d-%b-%Y %H:%M:%S GMT", t) - if isinstance(t, timedelta): + elif isinstance(t, timedelta): t = datetime.now() + t - if isinstance(t, datetime): + elif isinstance(t, datetime): d = t.utcoffset() if d is None: d = localTimeDelta() diff --git a/webware/JSONRPCServlet.py b/webware/JSONRPCServlet.py index 99249ea..a16b22f 100644 --- a/webware/JSONRPCServlet.py +++ b/webware/JSONRPCServlet.py @@ -83,12 +83,13 @@ def jsonCall(self): try: if self._debug: self.log(f"json call {call}(call)") - if isinstance(params, list): - result = method(*params) - elif isinstance(params, dict): - result = method(**params) - else: - result = method() + match params: + case list(): + result = method(*params) + case dict(): + result = method(**params) + case _: + result = method() self.writeResult(result) except Exception: err = StringIO() diff --git a/webware/MiscUtils/DBPool.py b/webware/MiscUtils/DBPool.py index 8440b21..5a4e947 100644 --- a/webware/MiscUtils/DBPool.py +++ b/webware/MiscUtils/DBPool.py @@ -128,6 +128,20 @@ def __init__(self, dbapi, maxconnections, *args, **kwargs): for _count in range(maxconnections): self.addConnection(dbapi.connect(*args, **kwargs)) + def close(self): + """Close all connections in the pool.""" + try: + queue = self._queue + except AttributeError: + connections = self._connections + while connections: + con = connections.pop() + con.close() + else: + while not queue.empty(): + con = queue.get_nowait() + con.close() + # The following functions are used with DB-API 2 modules # that do not have connection level threadsafety, like PyGreSQL. # However, the module must be threadsafe at the module level. diff --git a/webware/MiscUtils/DataTable.py b/webware/MiscUtils/DataTable.py index 1e79826..64cecfb 100644 --- a/webware/MiscUtils/DataTable.py +++ b/webware/MiscUtils/DataTable.py @@ -579,10 +579,12 @@ def setHeadings(self, headings): """ if not headings: self._headings = [] - elif isinstance(headings[0], str): - self._headings = list(map(TableColumn, headings)) - elif isinstance(headings[0], TableColumn): - self._headings = list(headings) + else: + match headings[0]: + case str(): + self._headings = list(map(TableColumn, headings)) + case TableColumn(): + self._headings = list(headings) for heading in self._headings: if heading.type() is None: heading.setType(self._defaultType) @@ -702,7 +704,7 @@ def __init__(self, table, values=None, headings=None): self._headings = table.headings() if headings is None else headings self._nameToIndexMap = table.nameToIndexMap() if values is not None: - if isinstance(values, (list, tuple)): + if isinstance(values, list | tuple): self.initFromSequence(values) elif isinstance(values, dict): self.initFromDict(values) @@ -791,8 +793,7 @@ def __repr__(self): return repr(self._values) def __iter__(self): - for value in self._values: - yield value + yield from self._values def get(self, key, default=None): index = self._nameToIndexMap.get(key) @@ -801,7 +802,7 @@ def get(self, key, default=None): return self._values[index] def has_key(self, key): - warn("has_key is deprecated, please us 'in' instead.", + warn("has_key is deprecated, use 'in' instead.", DeprecationWarning, stacklevel=2) return key in self diff --git a/webware/MiscUtils/DictForArgs.py b/webware/MiscUtils/DictForArgs.py index c660e73..6c72c23 100644 --- a/webware/MiscUtils/DictForArgs.py +++ b/webware/MiscUtils/DictForArgs.py @@ -193,11 +193,10 @@ def expandDictWithExtras(d, key='Extras', to specify attributes that occur infrequently. """ if key in d: - newDict = dict(d) + baseDict = dict(d) if delKey: - del newDict[key] - newDict.update(dictForArgs(d[key])) - return newDict + del baseDict[key] + return baseDict | dictForArgs(d[key]) return d diff --git a/webware/MiscUtils/M2PickleRPC.py b/webware/MiscUtils/M2PickleRPC.py index 03b1d00..188feb3 100644 --- a/webware/MiscUtils/M2PickleRPC.py +++ b/webware/MiscUtils/M2PickleRPC.py @@ -34,10 +34,7 @@ def make_connection(self, host, port=None): def parse_response(self, f): """Workaround M2Crypto issue mentioned above.""" sio = BytesIO() - while True: - chunk = f.read() - if not chunk: - break + while chunk := f.read(): sio.write(chunk) sio.seek(0) return Transport.parse_response(self, sio) @@ -45,10 +42,7 @@ def parse_response(self, f): def parse_response_gzip(self, f): """Workaround M2Crypto issue mentioned above.""" sio = BytesIO() - while True: - chunk = f.read() - if not chunk: - break + while chunk := f.read(): sio.write(chunk) sio.seek(0) return Transport.parse_response_gzip(self, sio) diff --git a/webware/MiscUtils/MixIn.py b/webware/MiscUtils/MixIn.py index 84bd020..91dc663 100644 --- a/webware/MiscUtils/MixIn.py +++ b/webware/MiscUtils/MixIn.py @@ -66,8 +66,7 @@ def foo(self): # Track the mix-ins made for a particular class attrName = 'mixInsFor' + pyClass.__name__ - mixIns = getattr(pyClass, attrName, None) - if mixIns is None: + if (mixIns := getattr(pyClass, attrName, None)) is None: mixIns = [] setattr(pyClass, attrName, mixIns) @@ -85,7 +84,7 @@ def foo(self): continue # built in or descriptor else: member = getattr(mixInClass, name) - if isinstance(member, (FunctionType, MethodType)): + if isinstance(member, FunctionType | MethodType): if mixInSuperMethods: if hasattr(pyClass, name): origMember = getattr(pyClass, name) diff --git a/webware/MiscUtils/NamedValueAccess.py b/webware/MiscUtils/NamedValueAccess.py index 78a92b8..9405cff 100644 --- a/webware/MiscUtils/NamedValueAccess.py +++ b/webware/MiscUtils/NamedValueAccess.py @@ -75,14 +75,11 @@ def valueForKey(obj, key, default=NoDefault): method = getattr(cls, key, None) if not method: underKey = '_' + key - method = getattr(cls, underKey, None) if cls else None - if not method: - attr = getattr(obj, key, NoDefault) - if attr is NoDefault: - attr = getattr(obj, underKey, NoDefault) - if attr is NoDefault and cls is not None: - getitem = getattr(cls, '__getitem__', None) - if getitem: + if not (method := getattr(cls, underKey, None) if cls else None): + if (attr := getattr(obj, key, NoDefault)) is NoDefault: + if ((attr := getattr(obj, underKey, NoDefault)) + is NoDefault and cls is not None): + if getitem := getattr(cls, '__getitem__', None): try: getitem(obj, key) except KeyError: diff --git a/webware/MiscUtils/PickleCache.py b/webware/MiscUtils/PickleCache.py index b4d213d..6ac6e77 100644 --- a/webware/MiscUtils/PickleCache.py +++ b/webware/MiscUtils/PickleCache.py @@ -47,10 +47,7 @@ import sys from time import sleep from pprint import pprint -try: - from pickle import load, dump, HIGHEST_PROTOCOL as maxPickleProtocol -except ImportError: - from pickle import load, dump, HIGHEST_PROTOCOL as maxPickleProtocol +from pickle import load, dump, HIGHEST_PROTOCOL as maxPickleProtocol verbose = False diff --git a/webware/MiscUtils/PickleRPC.py b/webware/MiscUtils/PickleRPC.py index 8457448..7f58d91 100644 --- a/webware/MiscUtils/PickleRPC.py +++ b/webware/MiscUtils/PickleRPC.py @@ -371,11 +371,11 @@ def parse_response_gzip(self, f): class SafeTransport(Transport): """Handle an HTTPS transaction to a Pickle-RPC server.""" - def make_connection(self, host, port=None, key_file=None, cert_file=None): + def make_connection(self, host, port=None, **kwargs): """Create an HTTPS connection object from a host descriptor.""" try: from http.client import HTTPSConnection except ImportError as e: raise NotImplementedError( "Your version of http.client doesn't support HTTPS") from e - return HTTPSConnection(host, port, key_file, cert_file) + return HTTPSConnection(host, port, **kwargs) diff --git a/webware/MiscUtils/Tests/BenchCSVParser.py b/webware/MiscUtils/Tests/BenchCSVParser.py index 7436b14..b391333 100644 --- a/webware/MiscUtils/Tests/BenchCSVParser.py +++ b/webware/MiscUtils/Tests/BenchCSVParser.py @@ -1,10 +1,7 @@ import sys import time from glob import glob -try: - from cProfile import Profile -except ImportError: - from profile import Profile +from cProfile import Profile from MiscUtils.CSVParser import CSVParser diff --git a/webware/MiscUtils/Tests/BenchDataTable.py b/webware/MiscUtils/Tests/BenchDataTable.py index 83eb271..c507bce 100644 --- a/webware/MiscUtils/Tests/BenchDataTable.py +++ b/webware/MiscUtils/Tests/BenchDataTable.py @@ -3,10 +3,7 @@ import sys import time from glob import glob -try: - from cProfile import Profile -except ImportError: - from profile import Profile +from cProfile import Profile from MiscUtils.DataTable import DataTable diff --git a/webware/MiscUtils/Tests/TestDBPool.py b/webware/MiscUtils/Tests/TestDBPool.py index 8259813..b28d710 100644 --- a/webware/MiscUtils/Tests/TestDBPool.py +++ b/webware/MiscUtils/Tests/TestDBPool.py @@ -6,12 +6,19 @@ class TestDBPool(unittest.TestCase): + def setUp(self): + self.pool = DBPool(sqlite3, 10, database=':memory:') + + def tearDown(self): + self.pool.close() + def testDbPool(self): - pool = DBPool(sqlite3, 10, database=':memory:') + query = "select 1 union select 2 union select 3 order by 1" + result = [(1,), (2,), (3,)] for _count in range(15): - con = pool.connection() + con = self.pool.connection() cursor = con.cursor() - cursor.execute("select 1 union select 2 union select 3 order by 1") + cursor.execute(query) rows = cursor.fetchall() - self.assertEqual(rows, [(1,), (2,), (3,)]) + self.assertEqual(rows, result) con.close() diff --git a/webware/PSP/BraceConverter.py b/webware/PSP/BraceConverter.py index da2f645..fb9f060 100644 --- a/webware/PSP/BraceConverter.py +++ b/webware/PSP/BraceConverter.py @@ -60,30 +60,31 @@ def parseLine(self, line, writer): self.line = self.line[match.end(1):] else: c = self.line[0] - if c == "'": - self.handleQuote("'", writer) - self.skipQuote(writer) - elif c == '"': - self.handleQuote('"', writer) - self.skipQuote(writer) - elif c == '{': - self.openBrace(writer) - elif c == '}': - self.closeBrace(writer) - elif c == ':': - self.openBlock(writer) - elif c == '#': - writer.printChars(self.line) - self.line = '' - else: - # should never get here - raise ValueError(f'Invalid character: {c!r}') + match c: + case "'": + self.handleQuote("'", writer) + self.skipQuote(writer) + case '"': + self.handleQuote('"', writer) + self.skipQuote(writer) + case '{': + self.openBrace(writer) + case '}': + self.closeBrace(writer) + case ':': + self.openBlock(writer) + case '#': + writer.printChars(self.line) + self.line = '' + case _: + # should never get here + raise ValueError(f'Invalid character: {c!r}') writer.printChars('\n') def openBlock(self, writer): """Open a new block.""" - match = self._reColonBrace.match(self.line) - if match and not self.dictLevel: + if ((match := self._reColonBrace.match(self.line)) + and not self.dictLevel): writer.printChars(':') writer.pushIndent() if match.group(1): diff --git a/webware/PSP/PSPParser.py b/webware/PSP/PSPParser.py index d255cd8..ee5769b 100644 --- a/webware/PSP/PSPParser.py +++ b/webware/PSP/PSPParser.py @@ -117,16 +117,17 @@ def checkDirective(self, handler, reader): reader.advance(len(match)) # parse the directive attr:val pair dictionary attrs = reader.parseTagAttributes() - if match == 'page': - checkAttributes('Page directive', attrs, ([], { - 'imports', 'extends', 'method', - 'isThreadSafe', 'isInstanceSafe', - 'indentType', 'indentSpaces', - 'gobbleWhitespace', 'formatter'})) - elif match == 'include': - checkAttributes('Include directive', attrs, (['file'], [])) - else: - raise PSPParserException(f'{match} directive not implemented') + match match: + case 'page': + checkAttributes('Page directive', attrs, ([], { + 'imports', 'extends', 'method', + 'isThreadSafe', 'isInstanceSafe', + 'indentType', 'indentSpaces', + 'gobbleWhitespace', 'formatter'})) + case 'include': + checkAttributes('Include directive', attrs, (['file'], [])) + case _: + raise PSPParserException(f'{match} directive not implemented') reader.skipSpaces() # skip to where we expect a close tag if reader.matches('%>'): reader.advance(2) # advance past it diff --git a/webware/PSP/ParseEventHandler.py b/webware/PSP/ParseEventHandler.py index d47e31a..30d2f98 100644 --- a/webware/PSP/ParseEventHandler.py +++ b/webware/PSP/ParseEventHandler.py @@ -190,21 +190,23 @@ def handleDirective(self, directive, start, stop, attrs): """Flush any template data and create a new DirectiveGenerator.""" self._parser.flushCharData(self.tmplStart, self.tmplStop) # big switch - if directive == 'page': - for h in attrs: - if h in self.directiveHandlers: - self.directiveHandlers[h](self, attrs[h], start, stop) - else: - raise ValueError(f'No page directive handler: {h}') - elif directive == 'include': - filename = attrs['file'] - encoding = attrs.get('encoding') - try: - self._reader.pushFile(filename, encoding) - except IOError as e: - raise IOError(f'PSP include file not found: {filename}') from e - else: - raise ValueError('Invalid directive: {directive}') + match directive: + case 'page': + for h in attrs: + if h in self.directiveHandlers: + self.directiveHandlers[h](self, attrs[h], start, stop) + else: + raise ValueError(f'No page directive handler: {h}') + case 'include': + filename = attrs['file'] + encoding = attrs.get('encoding') + try: + self._reader.pushFile(filename, encoding) + except IOError as e: + raise IOError( + f'PSP include file not found: {filename}') from e + case _: + raise ValueError('Invalid directive: {directive}') def handleScript(self, start, stop, attrs): """Handle scripting elements""" diff --git a/webware/PSP/ServletWriter.py b/webware/PSP/ServletWriter.py index 756067a..d7f607f 100644 --- a/webware/PSP/ServletWriter.py +++ b/webware/PSP/ServletWriter.py @@ -47,16 +47,17 @@ def setIndention(self): self._indent = '\t' if self._useTabs else self._indentSpaces def setIndentType(self, indentType): - if indentType == 'tabs': - self._useTabs = True - self.setIndention() - elif indentType == 'spaces': - self._useTabs = False - self.setIndention() - elif indentType == 'braces': - self._useTabs = False - self._useBraces = True - self.setIndention() + match indentType: + case 'tabs': + self._useTabs = True + self.setIndention() + case 'spaces': + self._useTabs = False + self.setIndention() + case 'braces': + self._useTabs = False + self._useBraces = True + self.setIndention() def close(self): pyCode = self._file.getvalue() diff --git a/webware/PickleRPCServlet.py b/webware/PickleRPCServlet.py index 1d36ddd..f9b0ad0 100644 --- a/webware/PickleRPCServlet.py +++ b/webware/PickleRPCServlet.py @@ -4,10 +4,7 @@ import traceback from time import time -try: - from pickle import dumps, PickleError -except ImportError: - from pickle import dumps, PickleError +from pickle import dumps, PickleError try: import zlib diff --git a/webware/PlugInLoader.py b/webware/PlugInLoader.py index e4ce6a0..952aced 100644 --- a/webware/PlugInLoader.py +++ b/webware/PlugInLoader.py @@ -1,4 +1,4 @@ -import pkg_resources +from importlib.metadata import entry_points from PlugIn import PlugIn @@ -46,7 +46,7 @@ def loadPlugIns(self, plugInNames=None, verbose=None): entryPoints = { entry_point.name: entry_point for entry_point - in pkg_resources.iter_entry_points('webware.plugins')} + in entry_points(group='webware.plugins')} plugIns = {} for name in plugInNames: diff --git a/webware/Properties.py b/webware/Properties.py index 976f03b..515f62d 100644 --- a/webware/Properties.py +++ b/webware/Properties.py @@ -1,10 +1,10 @@ name = 'Webware for Python' -version = (3, 0, 10) +version = (3, 1, 0) status = 'stable' -requiredPyVersion = (3, 6) +requiredPyVersion = (3, 10) synopsis = ( "Webware for Python is a time-tested" diff --git a/webware/RPCServlet.py b/webware/RPCServlet.py index 71bc595..e725e00 100644 --- a/webware/RPCServlet.py +++ b/webware/RPCServlet.py @@ -35,15 +35,15 @@ def resultForException(self, e, trans): """ # report exception back to server setting = trans.application().setting('RPCExceptionReturn') - if setting == 'occurred': - result = 'unhandled exception' - elif setting == 'exception': - result = str(e) - elif setting == 'traceback': - result = ''.join(traceback.format_exception(*sys.exc_info())) - else: - raise ValueError(f'Invalid setting: {setting!r}') - return result + match setting: + case 'occurred': + return 'unhandled exception' + case 'exception': + return str(e) + case 'traceback': + return ''.join(traceback.format_exception(*sys.exc_info())) + case _: + raise ValueError(f'Invalid setting: {setting!r}') @staticmethod def sendOK(contentType, contents, trans, contentEncoding=None): diff --git a/webware/Scripts/WebwareCLI.py b/webware/Scripts/WebwareCLI.py index 4d2faaa..28a119a 100644 --- a/webware/Scripts/WebwareCLI.py +++ b/webware/Scripts/WebwareCLI.py @@ -28,10 +28,11 @@ def main(args=None): args = parser.parse_args(args) command = args.command del args.command - if command == 'make': - make(args) - elif command == 'serve': - serve(args) + match command: + case 'make': + make(args) + case 'serve': + serve(args) if __name__ == '__main__': diff --git a/webware/Servlet.py b/webware/Servlet.py index 239ba67..bec8edd 100644 --- a/webware/Servlet.py +++ b/webware/Servlet.py @@ -155,8 +155,7 @@ def serverSidePath(self, path=None): if self._serverSidePath is None: self._serverSidePath = self._transaction.request().serverSidePath() if path: - if path.startswith('/'): - path = path[1:] + path = path.removeprefix('/') return os.path.normpath(os.path.join( os.path.dirname(self._serverSidePath), path)) return self._serverSidePath diff --git a/webware/SessionStore.py b/webware/SessionStore.py index 79941f8..514cf1c 100644 --- a/webware/SessionStore.py +++ b/webware/SessionStore.py @@ -120,7 +120,7 @@ def __iter__(self): def has_key(self, key): """Check whether the session store has a given key.""" - warn("has_key is deprecated, please us 'in' instead.", + warn("has_key is deprecated, use 'in' instead.", DeprecationWarning, stacklevel=2) return key in self diff --git a/webware/Tests/TestEndToEnd/AppTest.py b/webware/Tests/TestEndToEnd/AppTest.py index 52d44d5..5879649 100644 --- a/webware/Tests/TestEndToEnd/AppTest.py +++ b/webware/Tests/TestEndToEnd/AppTest.py @@ -1,6 +1,7 @@ """Test Webware Application via its WSGI interface""" import sys +import traceback from io import StringIO from os import chdir, getcwd @@ -31,7 +32,8 @@ def setUpClass(cls): cls.app = app cls.testApp = TestApp(app) except Exception as e: - error = str(e) or 'Could not create application' + error = (f"ERROR: {type(e).__name__}: {e}\n" + f"Full traceback:\n{traceback.format_exc()}") else: error = None finally: @@ -42,15 +44,15 @@ def setUpClass(cls): output = '' if error: raise RuntimeError( - 'Error setting up application:\n' + error + - '\nOutput was:\n' + output) + 'Error setting up application:\n' + f'{error}\nOutput was:\n{output}') if cls.catchOutput and not ( output.startswith('Webware for Python') and 'Running in development mode' in output and 'Loading context' in output): raise AssertionError( 'Application was not properly started.' - ' Output was:\n' + output) + f' Output was:\n{output}') @classmethod def tearDownClass(cls): @@ -68,8 +70,8 @@ def tearDownClass(cls): 'Application is shutting down...\n' 'Application has been successfully shut down.'): raise AssertionError( - 'Application was not properly shut down. Output was:\n' - + output) + 'Application was not properly shut down.' + f' Output was:\n{output}') def setUp(self): self.output = '' diff --git a/webware/Tests/TestEndToEnd/TestExamples.py b/webware/Tests/TestEndToEnd/TestExamples.py index fae6fe3..974de44 100644 --- a/webware/Tests/TestEndToEnd/TestExamples.py +++ b/webware/Tests/TestEndToEnd/TestExamples.py @@ -419,7 +419,7 @@ def testImageDemo(self): r = self.testApp.get('/ImageDemo?fmt=.png') self.assertEqual(r.status, '200 OK') self.assertEqual(r.content_type, 'image/png') - self.assertTrue(2000 < r.content_length < 2500) + self.assertTrue(2000 < r.content_length < 4500) self.assertTrue(r.body.startswith(b'\x89PNG\r\n\x1a\n')) def testDominateDemo(self): diff --git a/webware/Tests/TestEndToEnd/TestMakeApp.py b/webware/Tests/TestEndToEnd/TestMakeApp.py index 77e8773..4e8e76c 100644 --- a/webware/Tests/TestEndToEnd/TestMakeApp.py +++ b/webware/Tests/TestEndToEnd/TestMakeApp.py @@ -39,7 +39,7 @@ def testMakeHelp(self): expected.append('[-u USER] [-g GROUP]') expected.append('WORK_DIR') expected = ' '.join(expected) - self.assertTrue(output.startswith(expected)) + self.assertTrue(output.startswith(expected), output) def testMakeNewApp(self): output = self.runMake(['MyApp']) diff --git a/webware/Tests/TestEndToEnd/TestServer.py b/webware/Tests/TestEndToEnd/TestServer.py index 5ed1405..d70453d 100644 --- a/webware/Tests/TestEndToEnd/TestServer.py +++ b/webware/Tests/TestEndToEnd/TestServer.py @@ -120,8 +120,7 @@ def run(self): encoding='utf-8', stdout=PIPE, stderr=STDOUT) as p: outputStarted = False while True: - ret = p.poll() - if ret is not None: + if (ret := p.poll()) is not None: break line = p.stdout.readline().rstrip() if line: @@ -133,8 +132,7 @@ def run(self): self.pollQueue.put(ret) if ret is None: while True: - ret = p.poll() - if ret is not None: + if (ret := p.poll()) is not None: break line = p.stdout.readline().rstrip() self.outputQueue.put(line) @@ -206,8 +204,8 @@ def testExpectedServerOutput(self): self.compareOutput(self.output, expectedOutput) startLine = f'Webware for Python {self.version} Application' self.assertEqual(self.output[0], startLine) - self.assertTrue(not any( - 'ERROR' in line or 'WARNING' in line for line in self.output)) + self.assertFalse( + any('ERROR' in line or 'WARNING' in line for line in self.output)) def testStartPage(self): self.assertTrue(self.running) diff --git a/webware/Transaction.py b/webware/Transaction.py index 417dc93..951f4ee 100644 --- a/webware/Transaction.py +++ b/webware/Transaction.py @@ -197,8 +197,7 @@ def writeExceptionReport(self, handler): handler.writeAttrs(self, self._exceptionReportAttrNames) for name in self._exceptionReportAttrNames: - obj = getattr(self, '_' + name, None) - if obj: + if obj := getattr(self, '_' + name, None): try: obj.writeExceptionReport(handler) except Exception: diff --git a/webware/URLParser.py b/webware/URLParser.py index 34ab96e..20b8eec 100644 --- a/webware/URLParser.py +++ b/webware/URLParser.py @@ -266,8 +266,7 @@ def parse(self, trans, requestPath): context.append('') parts = [] while context: - contextName = '/'.join(context) - if contextName in self._contexts: + if (contextName := '/'.join(context)) in self._contexts: break parts.insert(0, context.pop()) if context: @@ -641,18 +640,17 @@ def parseInit(self, trans, requestPath): restPath = '' if nextPart in mod.urlRedirect: redirectTo = mod.urlRedirect[nextPart] - redirectPath = restPath elif '' in mod.urlRedirect: redirectTo = mod.urlRedirect[''] - redirectPath = restPath else: redirectTo = None - if redirectTo: + if redirectTo := (mod.urlRedirect.get(nextPart) or + mod.urlRedirect.get('')): if isinstance(redirectTo, str): fp = FileParser(os.path.join(self._path, redirectTo)) else: fp = redirectTo - return fp.parse(trans, redirectPath) + return fp.parse(trans, restPath) if 'SubParser' not in seen and hasattr(mod, 'SubParser'): seen.add('SubParser') diff --git a/webware/UnknownFileTypeServlet.py b/webware/UnknownFileTypeServlet.py index 09acdf1..bfa404f 100644 --- a/webware/UnknownFileTypeServlet.py +++ b/webware/UnknownFileTypeServlet.py @@ -78,8 +78,7 @@ def filename(self, trans): A subclass could override this in order to serve files from other disk locations based on some logic. """ - filename = getattr(self, '_serverSideFilename', None) - if filename is None: + if (filename := getattr(self, '_serverSideFilename', None)) is None: filename = trans.request().serverSidePath() self._serverSideFilename = filename # cache it return filename @@ -197,10 +196,10 @@ def serveContent(self, trans): fileSize, mtime = stat[6], stat[8] if debug: print('>> UnknownFileType.serveContent()') - print('>> filename =', filename) - print('>> size=', fileSize) - fileDict = fileCache.get(filename) - if fileDict is not None and mtime != fileDict['mtime']: + print(f'>> {filename=}') + print(f'>> {fileSize=}') + if ((fileDict := fileCache.get(filename)) is not None + and mtime != fileDict['mtime']): # Cache is out of date; clear it. if debug: print('>> changed, clearing cache') @@ -241,8 +240,8 @@ def serveContent(self, trans): print('>> sending directly') numBytesSent = 0 while numBytesSent < fileSize: - data = f.read(min(fileSize-numBytesSent, readBufferSize)) - if data == '': + if not (data := f.read( + min(fileSize-numBytesSent, readBufferSize))): break # unlikely, but safety first response.write(data) numBytesSent += len(data) diff --git a/webware/WebUtils/CGITraceback.py b/webware/WebUtils/CGITraceback.py index 122790c..1fd1272 100644 --- a/webware/WebUtils/CGITraceback.py +++ b/webware/WebUtils/CGITraceback.py @@ -37,8 +37,7 @@ def breaker(): def html(context=5, options=None): if options: - opt = DefaultOptions.copy() - opt.update(options) + opt = DefaultOptions | options else: opt = DefaultOptions @@ -126,32 +125,32 @@ def html(context=5, options=None): names = [] dotted = [0, []] - def tokeneater(): - if type_ == tokenize.OP and token == '.': - dotted[0] = 1 - if type_ == tokenize.NAME and token not in keyword.kwlist: - if dotted[0]: - dotted[0] = 0 - dotted[1].append(token) - if token not in names: - names.append(dotted[1][:]) - elif token not in names: - if token != 'self': - names.append(token) - dotted[1] = [token] - if type_ == tokenize.NEWLINE: - raise IndexError - def linereader(): nonlocal lineno - line = linecache.getline(filename, lineno) - line = line.encode('utf-8') + line = linecache.getline(filename, lineno).encode('utf-8') lineno += 1 return line + def tokeneater(type_, token): + match (type_, token): + case (tokenize.OP, '.'): + dotted[0] = 1 + case (tokenize.NAME, token) if token not in keyword.kwlist: + if dotted[0]: + dotted[0] = 0 + dotted[1].append(token) + if token not in names: + names.append(dotted[1][:]) + elif token not in names: + if token != 'self': + names.append(token) + dotted[1] = [token] + case (tokenize.NEWLINE, _): + raise IndexError + for type_, token, _start, _end, _line in tokenize.tokenize(linereader): try: - tokeneater() + tokeneater(type_, token) except IndexError: break diff --git a/webware/WebUtils/ExpansiveHTMLForException.py b/webware/WebUtils/ExpansiveHTMLForException.py index b754b67..ec08a2e 100644 --- a/webware/WebUtils/ExpansiveHTMLForException.py +++ b/webware/WebUtils/ExpansiveHTMLForException.py @@ -17,8 +17,7 @@ def expansiveHTMLForException(context=5, options=None): """Create expansive HTML for exceptions.""" if options: - opt = HTMLForExceptionOptions.copy() - opt.update(options) + opt = HTMLForExceptionOptions | options else: opt = HTMLForExceptionOptions return CGITraceback.html(context=context, options=opt) diff --git a/webware/WebUtils/FieldStorage.py b/webware/WebUtils/FieldStorage.py index 42a5a95..1bdd457 100644 --- a/webware/WebUtils/FieldStorage.py +++ b/webware/WebUtils/FieldStorage.py @@ -316,10 +316,8 @@ def read_urlencoded(self): 'keep_blank_values': self.keep_blank_values, 'strict_parsing': self.strict_parsing, 'encoding': self.encoding, 'errors': self.errors} - if self.max_num_fields is not None: # Python >= 3.8 - kwargs['max_num_fields'] = self.max_num_fields - if self.separator != '&': # Python >= 3.9.2 - kwargs['separator'] = self.separator + kwargs['max_num_fields'] = self.max_num_fields + kwargs['separator'] = self.separator query = parse_qsl(qs, **kwargs) # contrary to the standard library, we only add those fields # from the query string that do not already appear in the body @@ -344,10 +342,8 @@ def read_multi(self, environ, keep_blank_values, strict_parsing): 'keep_blank_values': self.keep_blank_values, 'strict_parsing': self.strict_parsing, 'encoding': self.encoding, 'errors': self.errors} - if self.max_num_fields is not None: # Python >= 3.8 - kwargs['max_num_fields'] = self.max_num_fields - if self.separator != '&': # Python >= 3.9.2 - kwargs['separator'] = self.separator + kwargs['max_num_fields'] = self.max_num_fields + kwargs['separator'] = self.separator query = parse_qsl(self.qs_on_post, **kwargs) self.list.extend(MiniFieldStorage(key, value) for key, value in query) @@ -419,10 +415,8 @@ def read_single(self): 'keep_blank_values': self.keep_blank_values, 'strict_parsing': self.strict_parsing, 'encoding': self.encoding, 'errors': self.errors} - if self.max_num_fields is not None: # Python >= 3.8 - kwargs['max_num_fields'] = self.max_num_fields - if self.separator != '&': # Python >= 3.9.2 - kwargs['separator'] = self.separator + kwargs['max_num_fields'] = self.max_num_fields + kwargs['separator'] = self.separator query = parse_qsl(self.qs_on_post, **kwargs) self.list = [MiniFieldStorage(key, value) for key, value in query] @@ -644,16 +638,6 @@ def valid_boundary(s): return pattern.match(s) -def hasSeparator(): - """Check whether the separator parameter is supported.""" - from urllib.parse import parse_qsl - try: - parse_qsl("", separator='&') - except TypeError: # Python < 3.9.2 - return False - return True - - def isBinaryType(ctype, pdict=None): """"Check whether the given MIME type uses binary data.""" if pdict and pdict.get('charset') == 'binary': diff --git a/webware/WebUtils/HTMLForException.py b/webware/WebUtils/HTMLForException.py index 9b46a21..936b58b 100644 --- a/webware/WebUtils/HTMLForException.py +++ b/webware/WebUtils/HTMLForException.py @@ -29,8 +29,7 @@ def htmlForLines(lines, options=None): # Set up the options: if options: - opt = HTMLForExceptionOptions.copy() - opt.update(options) + opt = HTMLForExceptionOptions | options else: opt = HTMLForExceptionOptions diff --git a/webware/WebUtils/HTMLTag.py b/webware/WebUtils/HTMLTag.py index 3a35dd1..7080612 100644 --- a/webware/WebUtils/HTMLTag.py +++ b/webware/WebUtils/HTMLTag.py @@ -144,7 +144,7 @@ def addChild(self, child): The child will be another tag or a string (CDATA). """ - if not isinstance(child, (str, HTMLTag)): + if not isinstance(child, str | HTMLTag): raise HTMLTagError(f'Invalid child: {child!r}') self._children.append(child) if isinstance(child, HTMLTag): diff --git a/webware/WebUtils/Tests/TestFieldStorageModified.py b/webware/WebUtils/Tests/TestFieldStorageModified.py index 39d238a..5d69d0d 100644 --- a/webware/WebUtils/Tests/TestFieldStorageModified.py +++ b/webware/WebUtils/Tests/TestFieldStorageModified.py @@ -4,7 +4,7 @@ from io import BytesIO -from WebUtils.FieldStorage import FieldStorage, hasSeparator, isBinaryType +from WebUtils.FieldStorage import FieldStorage, isBinaryType class TestFieldStorage(unittest.TestCase): @@ -85,52 +85,40 @@ def testPostRequestWithQueryWithSemicolon1(self): self.assertEqual(fs.getfirst('c'), '3') self.assertEqual(fs.getlist('a'), ['1']) self.assertEqual(fs.getlist('c'), ['3']) - if hasSeparator(): # new Python version, splits only & - self.assertEqual(fs.getfirst('b'), '2;b=3') - self.assertEqual(fs.getlist('b'), ['2;b=3']) - fs = FieldStorage( - fp=BytesIO(), - environ={'REQUEST_METHOD': 'POST', - 'QUERY_STRING': 'a=1&b=2&b=3&c=3'}, - separator='&') - self.assertEqual(fs.getfirst('a'), '1') - self.assertEqual(fs.getfirst('b'), '2') - self.assertEqual(fs.getfirst('c'), '3') - self.assertEqual(fs.getlist('a'), ['1']) - self.assertEqual(fs.getlist('b'), ['2', '3']) - self.assertEqual(fs.getlist('c'), ['3']) - else: # old Python version, splits ; and & - self.assertEqual(fs.getfirst('b'), '2') - self.assertEqual(fs.getlist('b'), ['2', '3']) + self.assertEqual(fs.getfirst('b'), '2;b=3') + self.assertEqual(fs.getlist('b'), ['2;b=3']) + fs = FieldStorage( + fp=BytesIO(), + environ={'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'a=1&b=2&b=3&c=3'}, + separator='&') + self.assertEqual(fs.getfirst('a'), '1') + self.assertEqual(fs.getfirst('b'), '2') + self.assertEqual(fs.getfirst('c'), '3') + self.assertEqual(fs.getlist('a'), ['1']) + self.assertEqual(fs.getlist('b'), ['2', '3']) + self.assertEqual(fs.getlist('c'), ['3']) def testPostRequestWithQueryWithSemicolon2(self): fs = FieldStorage(fp=BytesIO(), environ={ 'REQUEST_METHOD': 'POST', 'QUERY_STRING': 'a=1;b=2&b=3;c=3'}) - if hasSeparator(): # new Python version, splits only & - self.assertEqual(fs.getfirst('a'), '1;b=2') - self.assertEqual(fs.getfirst('b'), '3;c=3') - self.assertIsNone(fs.getfirst('c')) - self.assertEqual(fs.getlist('a'), ['1;b=2']) - self.assertEqual(fs.getlist('b'), ['3;c=3']) - self.assertEqual(fs.getlist('c'), []) - fs = FieldStorage( - fp=BytesIO(), - environ={'REQUEST_METHOD': 'POST', - 'QUERY_STRING': 'a=1;b=2;b=3;c=3'}, - separator=';') - self.assertEqual(fs.getfirst('a'), '1') - self.assertEqual(fs.getfirst('b'), '2') - self.assertEqual(fs.getfirst('c'), '3') - self.assertEqual(fs.getlist('a'), ['1']) - self.assertEqual(fs.getlist('b'), ['2', '3']) - self.assertEqual(fs.getlist('c'), ['3']) - else: # old Python version, splits ; and & - self.assertEqual(fs.getfirst('a'), '1') - self.assertEqual(fs.getfirst('b'), '2') - self.assertEqual(fs.getfirst('c'), '3') - self.assertEqual(fs.getlist('a'), ['1']) - self.assertEqual(fs.getlist('b'), ['2', '3']) - self.assertEqual(fs.getlist('c'), ['3']) + self.assertEqual(fs.getfirst('a'), '1;b=2') + self.assertEqual(fs.getfirst('b'), '3;c=3') + self.assertIsNone(fs.getfirst('c')) + self.assertEqual(fs.getlist('a'), ['1;b=2']) + self.assertEqual(fs.getlist('b'), ['3;c=3']) + self.assertEqual(fs.getlist('c'), []) + fs = FieldStorage( + fp=BytesIO(), + environ={'REQUEST_METHOD': 'POST', + 'QUERY_STRING': 'a=1;b=2;b=3;c=3'}, + separator=';') + self.assertEqual(fs.getfirst('a'), '1') + self.assertEqual(fs.getfirst('b'), '2') + self.assertEqual(fs.getfirst('c'), '3') + self.assertEqual(fs.getlist('a'), ['1']) + self.assertEqual(fs.getlist('b'), ['2', '3']) + self.assertEqual(fs.getlist('c'), ['3']) def testPostRequestWithoutContentLength(self): # see https://github.com/python/cpython/issues/71964 diff --git a/webware/WebUtils/Tests/TestFieldStorageStandard.py b/webware/WebUtils/Tests/TestFieldStorageStandard.py index 22f5ee2..8f3ed87 100644 --- a/webware/WebUtils/Tests/TestFieldStorageStandard.py +++ b/webware/WebUtils/Tests/TestFieldStorageStandard.py @@ -44,7 +44,6 @@ def testFieldStorageInvalid(self): fs = cgi.FieldStorage(headers={'content-type': 'text/plain'}) self.assertRaises(TypeError, bool, fs) - @unittest.skipUnless(cgi.hasSeparator(), "separator not supported") def testSeparator(self): parseSemicolon = [ ("x=1;y=2.0", {'x': ['1'], 'y': ['2.0']}), diff --git a/webware/__init__.py b/webware/__init__.py index f54e7c5..82b39e6 100644 --- a/webware/__init__.py +++ b/webware/__init__.py @@ -1,5 +1,9 @@ """Webware for Python""" +from .Properties import version as versionTuple + +__version__ = '.'.join(map(str, versionTuple)) + def addToSearchPath(): """Add the Webware package to the search path for Python modules."""