Source code for apache_beam.utils.annotations

#
# 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.
#

"""Deprecated annotations.

Deprecated: Signifies that users are discouraged from using a public API
typically because a better alternative exists, and the current form might be
removed in a future version.

Usage:
For internal use only; no backwards-compatibility guarantees.

Annotations come in two flavors: deprecated and experimental

The 'deprecated' annotation requires a 'since" parameter to specify
what version deprecated it.
Both 'deprecated' and 'experimental' annotations can specify the
current recommended version to use by means of a 'current' parameter.

The following example illustrates how to annotate coexisting versions of the
same function 'multiply'.::

  def multiply(arg1, arg2):
    print(arg1, '*', arg2, '=', end=' ')
    return arg1*arg2

# This annotation marks 'old_multiply' as deprecated since 'v.1' and suggests
# using 'multiply' instead.::

  @deprecated(since='v.1', current='multiply')
  def old_multiply(arg1, arg2):
    result = 0
    for i in xrange(arg1):
        result += arg2
    print(arg1, '*', arg2, '(the old way)=', end=' ')
    return result

# Set a warning filter to control how often warnings are produced.::

  warnings.simplefilter("always")
  print(multiply(5, 6))
  print(old_multiply(5,6))
"""

# pytype: skip-file

import inspect
import warnings
from functools import partial
from functools import wraps


[docs] class BeamDeprecationWarning(DeprecationWarning): """Beam-specific deprecation warnings."""
# Don't ignore BeamDeprecationWarnings. warnings.simplefilter('once', BeamDeprecationWarning) class _WarningMessage: """Utility class for assembling the warning message.""" def __init__(self, label, since, current, extra_message, custom_message): """Initialize message, leave only name as placeholder.""" if custom_message is None: message = '%name% is ' + label if label == 'deprecated': message += ' since %s' % since message += '. Use %s instead.' % current if current else '.' if extra_message: message += ' ' + extra_message else: if label == 'deprecated' and '%since%' not in custom_message: raise TypeError( "Replacement string %since% not found on \ custom message") emptyArg = lambda x: '' if x is None else x message = custom_message\ .replace('%since%', emptyArg(since))\ .replace('%current%', emptyArg(current))\ .replace('%extra%', emptyArg(extra_message)) self.label = label self.message = message def emit_warning(self, fnc_name): if self.label == 'deprecated': warning_type = BeamDeprecationWarning else: warning_type = FutureWarning warnings.warn( self.message.replace('%name%', fnc_name), warning_type, stacklevel=3)
[docs] def annotate(label, since, current, extra_message, custom_message=None): """Decorates an API with a deprecated or experimental annotation. Args: label: the kind of annotation ('deprecated' or 'experimental'). since: the version that causes the annotation. current: the suggested replacement function. extra_message: an optional additional message. custom_message: if the default message does not suffice, the message can be changed using this argument. A string whit replacement tokens. A replecement string is were the previus args will be located on the custom message. The following replacement strings can be used: %name% -> API.__name__ %since% -> since (Mandatory for the decapreted annotation) %current% -> current %extra% -> extra_message Returns: The decorator for the API. """ warning_message = _WarningMessage( label=label, since=since, current=current, extra_message=extra_message, custom_message=custom_message) def _annotate(fnc): if inspect.isclass(fnc): # Wrapping class into function causes documentation rendering issue. # Patch class's __new__ method instead. old_new = fnc.__new__ def wrapped_new(cls, *args, **kwargs): # Emit a warning when class instance is created. warning_message.emit_warning(fnc.__name__) if old_new is object.__new__: # object.__new__ takes no extra argument for python>=3 return old_new(cls) return old_new(cls, *args, **kwargs) fnc.__new__ = staticmethod(wrapped_new) return fnc else: @wraps(fnc) def inner(*args, **kwargs): # Emit a warning when the function is called. warning_message.emit_warning(fnc.__name__) return fnc(*args, **kwargs) return inner return _annotate
# Use partial application to customize each annotation. # 'current' will be optional in both deprecated and experimental # while 'since' will be mandatory for deprecated. deprecated = partial( annotate, label='deprecated', current=None, extra_message=None)