Code Refactor

I have changed code across several files to be more Pythonic, readable and succinct.
This commit is contained in:
HighnessAtharva 2023-01-05 00:27:04 +05:30
parent fa1ee70668
commit 2585f73ab4
8 changed files with 54 additions and 89 deletions

View file

@ -9,6 +9,7 @@ import weakref
from collections.abc import Iterator as _abc_Iterator from collections.abc import Iterator as _abc_Iterator
from datetime import timedelta from datetime import timedelta
from itertools import chain from itertools import chain
from itertools import product
from threading import Lock from threading import Lock
from types import TracebackType from types import TracebackType
@ -823,9 +824,7 @@ class Flask(Scaffold):
.. versionadded:: 0.8 .. versionadded:: 0.8
""" """
root_path = self.root_path root_path = self.instance_path if instance_relative else self.root_path
if instance_relative:
root_path = self.instance_path
defaults = dict(self.default_config) defaults = dict(self.default_config)
defaults["ENV"] = os.environ.get("FLASK_ENV") or "production" defaults["ENV"] = os.environ.get("FLASK_ENV") or "production"
defaults["DEBUG"] = get_debug_flag() defaults["DEBUG"] = get_debug_flag()
@ -994,7 +993,7 @@ class Flask(Scaffold):
for name in names: for name in names:
if name in self.template_context_processors: if name in self.template_context_processors:
for func in self.template_context_processors[name]: for func in self.template_context_processors[name]:
context.update(func()) context |= func()
context.update(orig_ctx) context.update(orig_ctx)
@ -1007,7 +1006,7 @@ class Flask(Scaffold):
""" """
rv = {"app": self, "g": g} rv = {"app": self, "g": g}
for processor in self.shell_context_processors: for processor in self.shell_context_processors:
rv.update(processor()) rv |= processor()
return rv return rv
@property @property
@ -1167,11 +1166,7 @@ class Flask(Scaffold):
sn_host, _, sn_port = server_name.partition(":") sn_host, _, sn_port = server_name.partition(":")
if not host: if not host:
if sn_host: host = sn_host or "127.0.0.1"
host = sn_host
else:
host = "127.0.0.1"
if port or port == 0: if port or port == 0:
port = int(port) port = int(port)
elif sn_port: elif sn_port:
@ -1559,18 +1554,17 @@ class Flask(Scaffold):
exc_class, code = self._get_exc_class_and_code(type(e)) exc_class, code = self._get_exc_class_and_code(type(e))
names = (*request.blueprints, None) names = (*request.blueprints, None)
for c in (code, None) if code is not None else (None,): for c, name in product((code, None) if code is not None else (None,), names):
for name in names: handler_map = self.error_handler_spec[name][c]
handler_map = self.error_handler_spec[name][c]
if not handler_map: if not handler_map:
continue continue
for cls in exc_class.__mro__: for cls in exc_class.__mro__:
handler = handler_map.get(cls) handler = handler_map.get(cls)
if handler is not None: if handler is not None:
return handler return handler
return None return None
def handle_http_exception( def handle_http_exception(
@ -1604,9 +1598,7 @@ class Flask(Scaffold):
return e return e
handler = self._find_error_handler(e) handler = self._find_error_handler(e)
if handler is None: return e if handler is None else self.ensure_sync(handler)(e)
return e
return self.ensure_sync(handler)(e)
def trap_http_exception(self, e: Exception) -> bool: def trap_http_exception(self, e: Exception) -> bool:
"""Checks if an HTTP exception should be trapped or not. By default """Checks if an HTTP exception should be trapped or not. By default
@ -1638,10 +1630,7 @@ class Flask(Scaffold):
): ):
return True return True
if trap_bad_request: return isinstance(e, BadRequest) if trap_bad_request else False
return isinstance(e, BadRequest)
return False
def handle_user_exception( def handle_user_exception(
self, e: Exception self, e: Exception
@ -1795,7 +1784,8 @@ class Flask(Scaffold):
): ):
return self.make_default_options_response() return self.make_default_options_response()
# otherwise dispatch to the handler for that endpoint # otherwise dispatch to the handler for that endpoint
view_args: t.Dict[str, t.Any] = req.view_args # type: ignore[assignment] # type: ignore[assignment]
view_args: t.Dict[str, t.Any] = req.view_args
return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args) return self.ensure_sync(self.view_functions[rule.endpoint])(**view_args)
def full_dispatch_request(self) -> Response: def full_dispatch_request(self) -> Response:
@ -1886,10 +1876,7 @@ class Flask(Scaffold):
.. versionadded:: 2.0 .. versionadded:: 2.0
""" """
if iscoroutinefunction(func): return self.async_to_sync(func) if iscoroutinefunction(func) else func
return self.async_to_sync(func)
return func
def async_to_sync( def async_to_sync(
self, func: t.Callable[..., t.Coroutine] self, func: t.Callable[..., t.Coroutine]
@ -2142,7 +2129,7 @@ class Flask(Scaffold):
# make sure the body is an instance of the response class # make sure the body is an instance of the response class
if not isinstance(rv, self.response_class): if not isinstance(rv, self.response_class):
if isinstance(rv, (str, bytes, bytearray)) or isinstance(rv, _abc_Iterator): if isinstance(rv, (str, bytes, bytearray, _abc_Iterator)):
# let the response class set the status and headers instead of # let the response class set the status and headers instead of
# waiting to do it manually, so that the class can handle any # waiting to do it manually, so that the class can handle any
# special logic # special logic
@ -2213,11 +2200,11 @@ class Flask(Scaffold):
# If subdomain matching is disabled (the default), use the # If subdomain matching is disabled (the default), use the
# default subdomain in all cases. This should be the default # default subdomain in all cases. This should be the default
# in Werkzeug but it currently does not have that feature. # in Werkzeug but it currently does not have that feature.
if not self.subdomain_matching: subdomain = (
subdomain = self.url_map.default_subdomain or None None
else: if self.subdomain_matching
subdomain = None else self.url_map.default_subdomain or None
)
return self.url_map.bind_to_environ( return self.url_map.bind_to_environ(
request.environ, request.environ,
server_name=self.config["SERVER_NAME"], server_name=self.config["SERVER_NAME"],

View file

@ -84,7 +84,7 @@ class BlueprintSetupState:
#: A dictionary with URL defaults that is added to each and every #: A dictionary with URL defaults that is added to each and every
#: URL that was defined with the blueprint. #: URL that was defined with the blueprint.
self.url_defaults = dict(self.blueprint.url_values_defaults) self.url_defaults = dict(self.blueprint.url_values_defaults)
self.url_defaults.update(self.options.get("url_defaults", ())) self.url_defaults |= self.options.get("url_defaults", ())
def add_url_rule( def add_url_rule(
self, self,
@ -387,7 +387,7 @@ class Blueprint(Scaffold):
f" provide a unique name." f" provide a unique name."
) )
first_bp_registration = not any(bp is self for bp in app.blueprints.values()) first_bp_registration = all(bp is not self for bp in app.blueprints.values())
first_name_registration = name not in app.blueprints first_name_registration = name not in app.blueprints
app.blueprints[name] = self app.blueprints[name] = self
@ -414,9 +414,7 @@ class Blueprint(Scaffold):
value = defaultdict( value = defaultdict(
dict, dict,
{ {
code: { code: dict(code_values.items())
exc_class: func for exc_class, func in code_values.items()
}
for code, code_values in value.items() for code, code_values in value.items()
}, },
) )

View file

@ -299,20 +299,17 @@ class ScriptInfo:
if self.create_app is not None: if self.create_app is not None:
app = self.create_app() app = self.create_app()
elif self.app_import_path:
path, name = (re.split(r":(?![\\/])", self.app_import_path, 1) + [None])[:2]
import_name = prepare_import(path)
app = locate_app(import_name, name)
else: else:
if self.app_import_path: for path in ("wsgi.py", "app.py"):
path, name = (
re.split(r":(?![\\/])", self.app_import_path, 1) + [None]
)[:2]
import_name = prepare_import(path) import_name = prepare_import(path)
app = locate_app(import_name, name) app = locate_app(import_name, None, raise_if_not_found=False)
else:
for path in ("wsgi.py", "app.py"):
import_name = prepare_import(path)
app = locate_app(import_name, None, raise_if_not_found=False)
if app: if app:
break break
if not app: if not app:
raise NoAppException( raise NoAppException(
@ -797,14 +794,14 @@ def _validate_key(ctx, param, value):
'When "--cert" is an SSLContext object, "--key is not used.', ctx, param 'When "--cert" is an SSLContext object, "--key is not used.', ctx, param
) )
if not cert: if cert:
ctx.params["cert"] = cert, value
else:
raise click.BadParameter('"--cert" must also be specified.', ctx, param) raise click.BadParameter('"--cert" must also be specified.', ctx, param)
ctx.params["cert"] = cert, value elif cert and not is_adhoc and not is_context:
raise click.BadParameter('Required when using "--cert".', ctx, param)
else:
if cert and not (is_adhoc or is_context):
raise click.BadParameter('Required when using "--cert".', ctx, param)
return value return value
@ -962,7 +959,7 @@ def shell_command() -> None:
with open(startup) as f: with open(startup) as f:
eval(compile(f.read(), startup, "exec"), ctx) eval(compile(f.read(), startup, "exec"), ctx)
ctx.update(current_app.make_shell_context()) ctx |= current_app.make_shell_context()
# Site, customize, or startup script can set a hook to call when # Site, customize, or startup script can set a hook to call when
# entering interactive mode. The default one sets up readline with # entering interactive mode. The default one sets up readline with
@ -1008,7 +1005,7 @@ def routes_command(sort: str, all_methods: bool) -> None:
ignored_methods = set(() if all_methods else ("HEAD", "OPTIONS")) ignored_methods = set(() if all_methods else ("HEAD", "OPTIONS"))
if sort in ("endpoint", "rule"): if sort in {"endpoint", "rule"}:
rules = sorted(rules, key=attrgetter(sort)) rules = sorted(rules, key=attrgetter(sort))
elif sort == "methods": elif sort == "methods":
rules = sorted(rules, key=lambda rule: sorted(rule.methods)) # type: ignore rules = sorted(rules, key=lambda rule: sorted(rule.methods)) # type: ignore

View file

@ -1,3 +1,4 @@
import contextlib
import errno import errno
import json import json
import os import os
@ -133,12 +134,8 @@ class Config(dict):
value = os.environ[key] value = os.environ[key]
try: with contextlib.suppress(Exception):
value = loads(value) value = loads(value)
except Exception:
# Keep the value as a string if loading failed.
pass
# Change to key.removeprefix(prefix) on Python >= 3.9. # Change to key.removeprefix(prefix) on Python >= 3.9.
key = key[len_prefix:] key = key[len_prefix:]
@ -325,10 +322,7 @@ class Config(dict):
for k, v in self.items(): for k, v in self.items():
if not k.startswith(namespace): if not k.startswith(namespace):
continue continue
if trim_namespace: key = k[len(namespace) :] if trim_namespace else k
key = k[len(namespace) :]
else:
key = k
if lowercase: if lowercase:
key = key.lower() key = key.lower()
rv[key] = v rv[key] = v

View file

@ -374,8 +374,8 @@ class RequestContext:
session_interface = self.app.session_interface session_interface = self.app.session_interface
self.session = session_interface.open_session(self.app, self.request) self.session = session_interface.open_session(self.app, self.request)
if self.session is None: if self.session is None:
self.session = session_interface.make_null_session(self.app) self.session = session_interface.make_null_session(self.app)
# Match the request URL after loading the session, so that the # Match the request URL after loading the session, so that the
# session is available in custom URL converters. # session is available in custom URL converters.

View file

@ -129,9 +129,7 @@ def explain_template_loading_attempts(app: Flask, template, attempts) -> None:
info.append(f"{idx + 1:5}: trying loader of {src_info}") info.append(f"{idx + 1:5}: trying loader of {src_info}")
for line in _dump_loader_info(loader): info.extend(f" {line}" for line in _dump_loader_info(loader))
info.append(f" {line}")
if triple is None: if triple is None:
detail = "no match" detail = "no match"
else: else:

View file

@ -28,7 +28,7 @@ class SessionMixin(MutableMapping):
@permanent.setter @permanent.setter
def permanent(self, value: bool) -> None: def permanent(self, value: bool) -> None:
self["_permanent"] = bool(value) self["_permanent"] = value
#: Some implementations can detect whether a session is newly #: Some implementations can detect whether a session is newly
#: created, but that is not guaranteed. Use with caution. The mixin #: created, but that is not guaranteed. Use with caution. The mixin
@ -195,7 +195,7 @@ class SessionInterface:
# set explicitly, or cached from SERVER_NAME detection # set explicitly, or cached from SERVER_NAME detection
# if False, return None # if False, return None
if rv is not None: if rv is not None:
return rv if rv else None return rv or None
rv = app.config["SERVER_NAME"] rv = app.config["SERVER_NAME"]

View file

@ -52,10 +52,7 @@ class Request(RequestBase):
@property @property
def max_content_length(self) -> t.Optional[int]: # type: ignore def max_content_length(self) -> t.Optional[int]: # type: ignore
"""Read-only view of the ``MAX_CONTENT_LENGTH`` config key.""" """Read-only view of the ``MAX_CONTENT_LENGTH`` config key."""
if current_app: return current_app.config["MAX_CONTENT_LENGTH"] if current_app else None
return current_app.config["MAX_CONTENT_LENGTH"]
else:
return None
@property @property
def endpoint(self) -> t.Optional[str]: def endpoint(self) -> t.Optional[str]:
@ -67,10 +64,7 @@ class Request(RequestBase):
This in combination with :attr:`view_args` can be used to This in combination with :attr:`view_args` can be used to
reconstruct the same URL or a modified URL. reconstruct the same URL or a modified URL.
""" """
if self.url_rule is not None: return self.url_rule.endpoint if self.url_rule is not None else None
return self.url_rule.endpoint
return None
@property @property
def blueprint(self) -> t.Optional[str]: def blueprint(self) -> t.Optional[str]:
@ -103,10 +97,7 @@ class Request(RequestBase):
""" """
name = self.blueprint name = self.blueprint
if name is None: return [] if name is None else _split_blueprint_path(name)
return []
return _split_blueprint_path(name)
def _load_form_data(self) -> None: def _load_form_data(self) -> None:
super()._load_form_data() super()._load_form_data()