"""Define Betty's core application functionality."""
from __future__ import annotations
import operator
import weakref
from concurrent.futures import Executor, ProcessPoolExecutor
from contextlib import suppress
from functools import reduce
from graphlib import CycleError, TopologicalSorter
from multiprocessing import get_context
from pathlib import Path
from types import TracebackType
from typing import TYPE_CHECKING, Mapping, Self, final
import aiohttp
from betty import fs
from betty.app.extension import ListExtensions, Extension, Extensions, build_extension_type_graph, \
CyclicDependencyError, ExtensionDispatcher, ConfigurableExtension, discover_extension_types
from betty.asyncio import sync, wait
from betty.cache import FileCache
from betty.config import Configurable, FileBasedConfiguration
from betty.dispatch import Dispatcher
from betty.fs import FileSystem, ASSETS_DIRECTORY_PATH
from betty.locale import LocalizerRepository, get_data, DEFAULT_LOCALE, Localizer, Str
from betty.model import Entity, EntityTypeProvider
from betty.model.event_type import EventType, EventTypeProvider, Birth, Baptism, Adoption, Death, Funeral, Cremation, \
Burial, Will, Engagement, Marriage, MarriageAnnouncement, Divorce, DivorceAnnouncement, Residence, Immigration, \
Emigration, Occupation, Retirement, Correspondence, Confirmation
from betty.project import Project
from betty.render import Renderer, SequentialRenderer
from betty.serde.dump import minimize, void_none, Dump, VoidableDump, Void
from betty.serde.load import AssertionFailed, Fields, Assertions, OptionalField, Asserter
if TYPE_CHECKING:
from betty.jinja2 import Environment
from betty.serve import Server
from betty.url import StaticUrlGenerator, LocalizedUrlGenerator
CONFIGURATION_DIRECTORY_PATH = fs.HOME_DIRECTORY_PATH / 'configuration'
class _AppExtensions(ListExtensions):
def __init__(self):
super().__init__([])
def _update(self, extensions: list[list[Extension]]) -> None:
self._extensions = extensions
[docs]
class AppConfiguration(FileBasedConfiguration):
def __init__(
self,
*,
locale: str | None = None,
):
super().__init__()
self._locale: str | None = locale
@property
def configuration_file_path(self) -> Path:
return CONFIGURATION_DIRECTORY_PATH / 'app.json'
@configuration_file_path.setter
def configuration_file_path(self, __) -> None:
pass
@configuration_file_path.deleter
def configuration_file_path(self) -> None:
pass
@property
def locale(self) -> str | None:
return self._locale
@locale.setter
def locale(self, locale: str) -> None:
try:
get_data(locale)
except ValueError:
raise AssertionFailed(Str._(
'"{locale}" is not a valid IETF BCP 47 language tag.',
locale=locale,
))
self._locale = locale
self._dispatch_change()
[docs]
def update(self, other: Self) -> None:
self._locale = other._locale
self._dispatch_change()
[docs]
@classmethod
def load(
cls,
dump: Dump,
configuration: Self | None = None,
) -> Self:
if configuration is None:
configuration = cls()
asserter = Asserter()
asserter.assert_record(Fields(
OptionalField(
'locale',
Assertions(asserter.assert_str()) | asserter.assert_setattr(configuration, 'locale')),
),
)(dump)
return configuration
[docs]
def dump(self) -> VoidableDump:
return minimize({
'locale': void_none(self.locale)
}, True)
[docs]
@final
class App(Configurable[AppConfiguration]):
def __init__(
self,
configuration: AppConfiguration | None = None,
project: Project | None = None,
):
super().__init__()
self._started = False
self._configuration = configuration or AppConfiguration()
self._configuration.on_change(self._on_locale_change)
self._assets: FileSystem | None = None
self._extensions = _AppExtensions()
self._extensions_initialized = False
self._localization_initialized = False
self._localizer: Localizer | None = None
self._localizers: LocalizerRepository | None = None
with suppress(FileNotFoundError):
wait(self.configuration.read())
self._project = project or Project()
self.project.configuration.extensions.on_change(self._update_extensions)
self._dispatcher: ExtensionDispatcher | None = None
self._entity_types: set[type[Entity]] | None = None
self._event_types: set[type[EventType]] | None = None
self._url_generator: LocalizedUrlGenerator | None = None
self._static_url_generator: StaticUrlGenerator | None = None
self._jinja2_environment: Environment | None = None
self._renderer: Renderer | None = None
self._http_client: aiohttp.ClientSession | None = None
self._cache: FileCache | None = None
self._process_pool: Executor | None = None
@classmethod
def _unreduce(cls, dumped_app_configuration: VoidableDump, project: Project) -> Self:
if dumped_app_configuration is Void:
app_configuration = None
else:
app_configuration = AppConfiguration.load(
dumped_app_configuration, # type: ignore[arg-type]
)
return App(
app_configuration,
project,
)
async def __aenter__(self) -> Self:
await self.start()
return self
async def __aexit__(self, exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None:
await self.stop()
[docs]
async def start(self) -> None:
if self._started:
raise RuntimeError('This app has started already.')
self._started = True
[docs]
async def stop(self) -> None:
self._started = False
del self.http_client
def __del__(self) -> None:
if self._started:
raise RuntimeError(f'{self} was started, but never stopped.')
def _on_locale_change(self) -> None:
del self.localizer
del self.localizers
@property
def project(self) -> Project:
return self._project
[docs]
def discover_extension_types(self) -> set[type[Extension]]:
return {*discover_extension_types(), *map(type, self._extensions.flatten())}
@property
def extensions(self) -> Extensions:
if not self._extensions_initialized:
self._extensions_initialized = True
self._update_extensions()
return self._extensions
def _update_extensions(self) -> None:
extension_types_enabled_in_configuration = set()
for app_extension_configuration in self.project.configuration.extensions.values():
if app_extension_configuration.enabled:
app_extension_configuration.extension_type.enable_requirement().assert_met()
extension_types_enabled_in_configuration.add(app_extension_configuration.extension_type)
extension_types_sorter = TopologicalSorter(
build_extension_type_graph(extension_types_enabled_in_configuration)
)
try:
extension_types_sorter.prepare()
except CycleError:
raise CyclicDependencyError([
app_extension_configuration.extension_type
for app_extension_configuration
in self.project.configuration.extensions.values()
])
extensions = []
while extension_types_sorter.is_active():
extension_types_batch = extension_types_sorter.get_ready()
extensions_batch = []
for extension_type in extension_types_batch:
if issubclass(extension_type, ConfigurableExtension) and extension_type in self.project.configuration.extensions:
extension: Extension = extension_type(self, configuration=self.project.configuration.extensions[extension_type].extension_configuration)
else:
extension = extension_type(self)
extensions_batch.append(extension)
extension_types_sorter.done(extension_type)
extensions.append(extensions_batch)
self._extensions._update(extensions)
del self.assets
del self.localizers
del self.localizer
del self.jinja2_environment
del self.renderer
del self.entity_types
del self.event_types
@property
def assets(self) -> FileSystem:
if self._assets is None:
assets = FileSystem()
assets.prepend(ASSETS_DIRECTORY_PATH, 'utf-8')
for extension in self.extensions.flatten():
extension_assets_directory_path = extension.assets_directory_path()
if extension_assets_directory_path is not None:
assets.prepend(extension_assets_directory_path, 'utf-8')
assets.prepend(self.project.configuration.assets_directory_path)
self._assets = assets
return self._assets
@assets.deleter
def assets(self) -> None:
self._assets = None
@property
def dispatcher(self) -> Dispatcher:
if self._dispatcher is None:
self._dispatcher = ExtensionDispatcher(self.extensions)
return self._dispatcher
@property
def url_generator(self) -> LocalizedUrlGenerator:
from betty.url import AppUrlGenerator
if self._url_generator is None:
self._url_generator = AppUrlGenerator(self)
return self._url_generator
@property
def static_url_generator(self) -> StaticUrlGenerator:
from betty.url import StaticPathUrlGenerator
if self._static_url_generator is None:
self._static_url_generator = StaticPathUrlGenerator(self.project.configuration)
return self._static_url_generator
@property
def localizer(self) -> Localizer:
"""
Get the application's localizer.
"""
if self._localizer is None:
self._localizer = wait(self.localizers.get_negotiated(self.configuration.locale or DEFAULT_LOCALE))
return self._localizer
@localizer.deleter
def localizer(self) -> None:
self._localizer = None
@property
def localizers(self) -> LocalizerRepository:
if self._localizers is None:
self._localizers = LocalizerRepository(self.assets)
return self._localizers
@localizers.deleter
def localizers(self) -> None:
self._localizers = None
@property
def jinja2_environment(self) -> Environment:
if not self._jinja2_environment:
from betty.jinja2 import Environment
self._jinja2_environment = Environment(self)
return self._jinja2_environment
@jinja2_environment.deleter
def jinja2_environment(self) -> None:
self._jinja2_environment = None
@property
def renderer(self) -> Renderer:
if not self._renderer:
from betty.jinja2 import Jinja2Renderer
self._renderer = SequentialRenderer([
Jinja2Renderer(self.jinja2_environment, self.project.configuration),
])
return self._renderer
@renderer.deleter
def renderer(self) -> None:
self._renderer = None
@property
def http_client(self) -> aiohttp.ClientSession:
if not self._http_client:
self._http_client = aiohttp.ClientSession(
connector=aiohttp.TCPConnector(limit_per_host=5),
headers={
'User-Agent': f'Betty (https://github.com/bartfeenstra/betty) on behalf of {self._project.configuration.base_url}{self._project.configuration.root_path}',
},
)
weakref.finalize(self, sync(self._http_client.close))
return self._http_client
@http_client.deleter
def http_client(self) -> None:
if self._http_client is not None:
wait(self._http_client.close())
self._http_client = None
@property
def entity_types(self) -> set[type[Entity]]:
if self._entity_types is None:
from betty.model.ancestry import Citation, Enclosure, Event, File, Note, Person, PersonName, Presence, Place, Source
self._entity_types = reduce(operator.or_, wait(self.dispatcher.dispatch(EntityTypeProvider)()), set()) | {
Citation,
Enclosure,
Event,
File,
Note,
Person,
PersonName,
Presence,
Place,
Source,
}
return self._entity_types
@entity_types.deleter
def entity_types(self) -> None:
self._entity_types = None
@property
def event_types(self) -> set[type[EventType]]:
if self._event_types is None:
self._event_types = set(wait(self.dispatcher.dispatch(EventTypeProvider)())) | {
Birth,
Baptism,
Adoption,
Death,
Funeral,
Cremation,
Burial,
Will,
Engagement,
Marriage,
MarriageAnnouncement,
Divorce,
DivorceAnnouncement,
Residence,
Immigration,
Emigration,
Occupation,
Retirement,
Correspondence,
Confirmation,
}
return self._event_types
@event_types.deleter
def event_types(self) -> None:
self._event_types = None
@property
def servers(self) -> Mapping[str, Server]:
from betty import serve
from betty.extension.demo import DemoServer
return {
server.name(): server
for server
in [
*(
server
for extension in self.extensions.flatten()
if isinstance(extension, serve.ServerProvider)
for server in extension.servers
),
serve.BuiltinAppServer(self),
DemoServer(),
]
}
@property
def cache(self) -> FileCache:
if self._cache is None:
self._cache = FileCache(self.localizer, fs.CACHE_DIRECTORY_PATH)
return self._cache
@property
def process_pool(self) -> Executor:
if self._process_pool is None:
# Avoid `fork` so as not to start worker processes with unneeded resources.
# Settle for `spawn` so all environments use the same start method.
self._process_pool = ProcessPoolExecutor(mp_context=get_context('spawn'))
return self._process_pool