Source code for forml.application._strategy

# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

"""
ForML application model rollout strategy.
"""
import abc
import typing

from forml.io import asset as assetmod

if typing.TYPE_CHECKING:
    from forml import runtime
    from forml.io import asset  # pylint: disable=reimported


[docs]class Selector(abc.ABC): """Abstract base class for the model selection strategy to be used by the :class:`application.Generic <forml.application.Generic>` descriptors. """
[docs] @abc.abstractmethod def select(self, registry: 'asset.Directory', context: typing.Any, stats: 'runtime.Stats') -> 'asset.Instance': """Select the model instance to be used for serving the request. See Also: This serves the same purpose as the :meth:`application.Descriptor.select <forml.application.Descriptor.select>` method only extracted as a separate object. Args: registry: Model registry to select the model from. context: Optional metadata carried over from the :meth:`application.Descriptor.receive <forml.application.Descriptor.receive>`. stats: Application specific serving metrics. Returns: Model instance. """
[docs]class Explicit(Selector): """Model selection strategy always choosing an explicit model generation. Args: project: Project reference of the selected model. release: Project release reference of the selected model. generation: Project generation reference of the selected model. """ def __init__( self, project: typing.Union[str, 'asset.Project.Key'], release: typing.Union[str, 'asset.Release.Key'], generation: typing.Union[str, int, 'asset.Generation.Key'], ): self._project: typing.Union[str, 'asset.Project.Key'] = project self._release: typing.Union[str, 'asset.Release.Key'] = release self._generation: typing.Union[str, int, 'asset.Generation.Key'] = generation self._instance: typing.Optional['asset.Instance'] = None def select(self, registry: 'asset.Directory', context: typing.Any, stats: 'runtime.Stats') -> 'asset.Instance': if not self._instance: self._instance = assetmod.Instance( registry=registry, project=self._project, release=self._release, generation=self._generation, ) return self._instance
[docs]class Latest(Selector): """Model selection strategy choosing an instance of the most recent model release/generation. Attention: Currently, the instance is cached indefinitely and so updates to the registry are not dynamically reflected. Args: project: Project reference to choose the most recent generation from. release: Optional release to choose the most recent generation from. """ def __init__( self, project: typing.Union[str, 'asset.Project.Key'], release: typing.Optional[typing.Union[str, 'asset.Release.Key']] = None, ): self._project: typing.Union[str, 'asset.Project.Key'] = project self._release: typing.Optional[typing.Union[str, 'asset.Release.Key']] = release self._instance: typing.Optional['asset.Instance'] = None def select(self, registry: 'asset.Directory', context: typing.Any, stats: 'runtime.Stats') -> 'asset.Instance': if not self._instance: release = self._release generation = None if not release: project = registry.get(self._project) for release in reversed(project.list()): try: generation = project.get(release).list().last except assetmod.Level.Listing.Empty: continue break else: raise assetmod.Level.Listing.Empty(f'No models available for {self._project}') self._instance = assetmod.Instance( registry=registry, project=self._project, release=release, # pylint: disable=undefined-loop-variable generation=generation, ) return self._instance