# Released under the MIT License. See LICENSE for details.
#
"""Generates parts of babase._app.py.
This includes things like subsystem attributes for all feature-sets that
want them and default app-intent handling.
"""
from __future__ import annotations
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from batools.featureset import FeatureSet
[docs]
def generate_app_module(
projroot: str, feature_sets: dict[str, FeatureSet], existing_data: str
) -> str:
"""Generate babase._app.py based on its existing version."""
# pylint: disable=too-many-locals
# pylint: disable=too-many-branches
# pylint: disable=too-many-statements
import textwrap
from efrotools.util import replace_section
from efrotools.project import getprojectconfig
out = ''
fsets = feature_sets
out = existing_data
info = f'# This section generated by {__name__}; do not edit.'
indent = ' '
# Import modules we need for feature-set subsystems.
contents = ''
for _fsname, fset in sorted(fsets.items()):
if fset.has_python_app_subsystem:
modname = fset.name_python_package
classname = f'{fset.name_camel}AppSubsystem'
contents += f'from {modname} import {classname}\n'
out = replace_section(
out,
f'{indent}# __FEATURESET_APP_SUBSYSTEM_IMPORTS_BEGIN__\n',
f'{indent}# __FEATURESET_APP_SUBSYSTEM_IMPORTS_END__\n',
textwrap.indent(
f'{info}\n\n{contents}\n' if contents else f'{info}\n', indent
),
keep_markers=True,
)
# Calc which feature-sets are soft-required by any of the ones here
# but not included here. For those we'll expose app-subsystems that
# always return None.
missing_soft_fset_names = set[str]()
for fset in fsets.values():
for softreq in fset.soft_requirements:
if softreq not in fsets:
missing_soft_fset_names.add(softreq)
all_fset_names = missing_soft_fset_names | fsets.keys()
# Add properties to instantiate feature-set subsystems.
contents = ''
for fsetname in sorted(all_fset_names):
if fsetname in missing_soft_fset_names:
contents += (
f'\n'
f'@property\n'
f'def {fsetname}(self) -> Any | None:\n'
f' """Our {fsetname} subsystem (not available'
f' in this project)."""\n'
f'\n'
f' return None\n'
)
else:
fset = fsets[fsetname]
if fset.has_python_app_subsystem:
modname = fset.name_python_package
classname = f'{fset.name_camel}AppSubsystem'
# If they are allowed as a soft requirement, *everyone*
# has to access them as TYPE | None. Originally I planned to
# add the '| None' *only* if another present feature set was
# soft referencing them, but it turns out that code tuned
# to expect TYPE hits a lot of 'code will never be executed'
# errors in the type checker if we switch it to 'TYPE | None'
# so we need to be consistent.
if fset.allow_as_soft_requirement:
contents += (
f'\n'
f'@property\n'
f'def {fset.name}(self) -> {classname} | None:\n'
f' """Our {fset.name} subsystem (if available)."""\n'
f' return self._get_subsystem_property(\n'
f" '{fset.name}', "
f'self._create_{fset.name}_subsystem\n'
f' ) # type: ignore\n'
f'\n'
f'@staticmethod\n'
f'def _create_{fset.name}_subsystem()'
f' -> {classname} | None:\n'
f' # pylint: disable=cyclic-import\n'
f' try:\n'
f' from {modname} import {classname}\n'
f'\n'
f' return {classname}()\n'
f' except ImportError:\n'
f' return None\n'
f' except Exception:\n'
f" logging.exception('Error importing"
f" {modname}.')\n"
f' return None\n'
)
else:
contents += (
f'\n'
'@property\n'
f'def {fset.name}(self) -> {classname}:\n'
f' """Our {fset.name} subsystem'
' (always available)."""\n'
f' return self._get_subsystem_property(\n'
f" '{fset.name}', "
f'self._create_{fset.name}_subsystem\n'
f' ) # type: ignore\n'
f'\n'
f'@staticmethod\n'
f'def _create_{fset.name}_subsystem()'
f' -> {classname}:\n'
f' # pylint: disable=cyclic-import\n'
f'\n'
f' from {modname} import {classname}\n'
f'\n'
f' return {classname}()\n'
)
out = replace_section(
out,
f'{indent}# __FEATURESET_APP_SUBSYSTEM_PROPERTIES_BEGIN__\n',
f'{indent}# __FEATURESET_APP_SUBSYSTEM_PROPERTIES_END__\n',
textwrap.indent(f'{info}\n{contents}\n', indent),
keep_markers=True,
)
# Generate code to create app subsystems in the proper order.
all_ss_fsets = {
fsetname: fset
for fsetname, fset in fsets.items()
if fset.has_python_app_subsystem
}
init_order: list[str] = []
for fsetname, fset in sorted(all_ss_fsets.items()):
_add_init(fset, all_ss_fsets, init_order, 0)
contents = '# Poke these attrs to create all our subsystems.\n' + ''.join(
f'_ = self.{fsetname}\n' for fsetname in init_order
)
indent = ' '
out = replace_section(
out,
f'{indent}# __FEATURESET_APP_SUBSYSTEM_CREATE_BEGIN__\n',
f'{indent}# __FEATURESET_APP_SUBSYSTEM_CREATE_END__\n',
textwrap.indent(f'{info}\n\n{contents}\n', indent),
keep_markers=True,
)
# Generate default app-mode-selection logic.
# TODO - make this customizable via project settings or whatnot.
default_app_modes: list[str] | None = getprojectconfig(projroot).get(
'default_app_modes'
)
if not isinstance(default_app_modes, list) or not all(
isinstance(x, str) for x in default_app_modes
):
raise RuntimeError(
'Could not load default_app_modes from projectconfig'
)
def _module_for_app_mode(amode: str) -> str:
return '.'.join(amode.split('.')[:-1])
def _is_valid_app_mode(amode: str) -> bool:
# Consider the app mode valid if it comes from a Python
# package provided by one of our feature-sets.
module = _module_for_app_mode(amode)
for featureset in feature_sets.values():
if featureset.name_python_package == module:
return True
return False
default_app_modes = [m for m in default_app_modes if _is_valid_app_mode(m)]
contents = (
'# Ask our default app modes to handle it.\n'
"# (generated from 'default_app_modes' in projectconfig).\n"
)
if not default_app_modes:
raise RuntimeError('No valid default_app_modes specified.')
for mode in default_app_modes:
contents += f'import {_module_for_app_mode(mode)}\n'
contents += '\n'
contents += 'for appmode in [\n'
for mode in default_app_modes:
contents += f' {mode},\n'
contents += (
']:\n'
' if appmode.can_handle_intent(intent):\n'
' return appmode\n'
'\n'
)
contents += 'return None\n'
indent = ' '
out = replace_section(
out,
f'{indent}# __DEFAULT_APP_MODE_SELECTION_BEGIN__\n',
f'{indent}# __DEFAULT_APP_MODE_SELECTION_END__\n',
textwrap.indent(f'{info}\n\n{contents}\n', indent),
keep_markers=True,
)
# Disabling this for now; will probably remove permanently. Testable
# app mode discovery now uses meta discovery system.
if bool(False):
contents = (
'# Return all our default_app_modes as testable.\n'
"# (generated from 'default_app_modes' in projectconfig).\n"
)
for mode in default_app_modes:
contents += f'import {_module_for_app_mode(mode)}\n'
contents += '\n'
contents += 'return [\n'
for mode in default_app_modes:
contents += f' {mode},\n'
contents += ']'
out = replace_section(
out,
f'{indent}# __DEFAULT_TESTABLE_APP_MODES_BEGIN__\n',
f'{indent}# __DEFAULT_TESTABLE_APP_MODES_END__\n',
textwrap.indent(f'{info}\n\n{contents}\n', indent),
keep_markers=True,
)
# Note: we *should* format this string, but because this code
# runs with every project update I'm just gonna try to keep the
# formatting correct manually for now to save a bit of time.
# (project update time jumps from 0.3 to 0.5 seconds if I enable
# formatting here for just this one file).
return out
def _add_init(
fset: FeatureSet,
allsets: dict[str, FeatureSet],
init_order: list[str],
depth: int,
) -> None:
# If we hit max recursion, we've got a dependency cycle.
if depth > 10:
raise RuntimeError(
'App subsystem dependency cycle detected'
f" (involving feature set '{fset.name}')."
)
# If this one is already added, we're done.
if fset.name in init_order:
return
# First add all of its dependencies.
for depname in sorted(fset.python_app_subsystem_dependencies):
depfset = allsets.get(depname)
# Only matters if this is in the actual set of featuresets.
if depfset is None:
continue
_add_init(depfset, allsets, init_order, depth + 1)
# We should not have been added via the above code (dependency cycle
# should have been detected in that case).
assert fset.name not in init_order
# Finally add the fset itself.
init_order.append(fset.name)
# Docs-generation hack; import some stuff that we likely only forward-declared
# in our actual source code so that docs tools can find it.
from typing import (Coroutine, Any, Literal, Callable,
Generator, Awaitable, Sequence, Self)
import asyncio
from concurrent.futures import Future
from pathlib import Path
from enum import Enum