azure/functions/_abc.py (401 lines of code) (raw):
# Copyright (c) Microsoft Corporation. All rights reserved.
# Licensed under the MIT License.
import abc
import datetime
import io
import threading
import typing
from werkzeug.datastructures import Headers
T = typing.TypeVar('T')
class Out(abc.ABC, typing.Generic[T]):
"""An interface to set function output parameters."""
@abc.abstractmethod
def set(self, val: T) -> None:
"""Set the value of the output parameter."""
pass
@abc.abstractmethod
def get(self) -> T:
"""Get the value of the output parameter."""
pass
class RpcException:
"""Rpc Exception object."""
@property
@abc.abstractmethod
def source(self) -> str:
"""Source of the exception."""
pass
@property
@abc.abstractmethod
def stack_trace(self) -> str:
"""Stack trace for the exception."""
pass
@property
@abc.abstractmethod
def message(self) -> str:
"""Textual message describing the exception."""
pass
class TraceContext(abc.ABC):
"""Trace context object."""
@property
@abc.abstractmethod
def trace_state(self) -> str:
"""Gets trace state from trace-context."""
pass
@property
@abc.abstractmethod
def trace_parent(self) -> str:
"""Gets trace parent from trace-context."""
pass
@property
@abc.abstractmethod
def attributes(self) -> typing.Dict[str, str]:
"""Gets trace-context attributes."""
pass
class RetryContext(abc.ABC):
"""Retry Context object.
For more information refer: https://aka.ms/azfunc-retries-policies
"""
@property
@abc.abstractmethod
def retry_count(self) -> int:
"""Gets the current retry count from retry-context."""
pass
@property
@abc.abstractmethod
def max_retry_count(self) -> int:
"""Gets the max retry count from retry-context."""
pass
@property
@abc.abstractmethod
def exception(self) -> RpcException:
"""Gets the RpcException"""
pass
class Context(abc.ABC):
"""Function invocation context."""
@property
@abc.abstractmethod
def invocation_id(self) -> str:
"""Function invocation ID."""
pass
@property
@abc.abstractmethod
def thread_local_storage(self) -> typing.Type[threading.local]:
"""Thread local storage.
:attribute str invocation_id:
Invocation ID contained in local thread storage.
Enables logging from user threads when set to
the current context's invocation ID.
"""
pass
@property
@abc.abstractmethod
def function_name(self) -> str:
"""Function name."""
pass
@property
@abc.abstractmethod
def function_directory(self) -> str:
"""Function directory."""
pass
@property
@abc.abstractmethod
def trace_context(self) -> TraceContext:
"""Context for distributed tracing."""
pass
@property
@abc.abstractmethod
def retry_context(self) -> RetryContext:
"""Context for retries to the function."""
pass
class HttpRequest(abc.ABC):
"""HTTP request object."""
@property
@abc.abstractmethod
def method(self) -> str:
"""Request method."""
pass
@property
@abc.abstractmethod
def url(self) -> str:
"""Request URL."""
pass
@property
@abc.abstractmethod
def headers(self) -> typing.Mapping[str, str]:
"""A dictionary containing request headers."""
pass
@property
@abc.abstractmethod
def params(self) -> typing.Mapping[str, str]:
"""A dictionary containing request GET parameters."""
pass
@property
@abc.abstractmethod
def route_params(self) -> typing.Mapping[str, str]:
"""A dictionary containing request route parameters."""
pass
@abc.abstractmethod
def get_body(self) -> bytes:
"""Return request body as bytes."""
pass
@abc.abstractmethod
def get_json(self) -> typing.Any:
"""Decode and return request body as JSON.
:raises ValueError:
when the request does not contain valid JSON data.
"""
pass
class HttpResponse(abc.ABC):
@property
@abc.abstractmethod
def status_code(self) -> int:
pass
@property
@abc.abstractmethod
def mimetype(self):
pass
@property
@abc.abstractmethod
def charset(self):
pass
@property
@abc.abstractmethod
def headers(self) -> Headers:
pass
@abc.abstractmethod
def get_body(self) -> bytes:
pass
class TimerRequest(abc.ABC):
"""Timer request object."""
@property
@abc.abstractmethod
def past_due(self) -> bool:
"""Whether the timer is past due."""
pass
class InputStream(io.BufferedIOBase, abc.ABC):
"""File-like object representing an input blob."""
@abc.abstractmethod
def read(self, size=-1) -> bytes:
"""Return and read up to *size* bytes.
:param int size:
The number of bytes to read. If the argument is omitted,
``None``, or negative, data is read and returned until
EOF is reached.
:return:
Bytes read from the input stream.
"""
pass
@property
@abc.abstractmethod
def name(self) -> typing.Optional[str]:
"""The name of the blob."""
pass
@property
@abc.abstractmethod
def length(self) -> typing.Optional[int]:
"""The size of the blob in bytes."""
pass
@property
@abc.abstractmethod
def uri(self) -> typing.Optional[str]:
"""The blob's primary location URI."""
pass
class QueueMessage(abc.ABC):
@property
@abc.abstractmethod
def id(self) -> typing.Optional[str]:
pass
@abc.abstractmethod
def get_body(self) -> typing.Union[str, bytes]:
pass
@abc.abstractmethod
def get_json(self) -> typing.Any:
pass
@property
@abc.abstractmethod
def dequeue_count(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def expiration_time(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def insertion_time(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def time_next_visible(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def pop_receipt(self) -> typing.Optional[str]:
pass
class EventGridEvent(abc.ABC):
@property
@abc.abstractmethod
def id(self) -> str:
pass
@abc.abstractmethod
def get_json(self) -> typing.Any:
pass
@property
@abc.abstractmethod
def topic(self) -> str:
pass
@property
@abc.abstractmethod
def subject(self) -> str:
pass
@property
@abc.abstractmethod
def event_type(self) -> str:
pass
@property
@abc.abstractmethod
def event_time(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def data_version(self) -> str:
pass
class EventGridOutputEvent(abc.ABC):
@property
@abc.abstractmethod
def id(self) -> str:
pass
@abc.abstractmethod
def get_json(self) -> typing.Any:
pass
@property
@abc.abstractmethod
def subject(self) -> str:
pass
@property
@abc.abstractmethod
def event_type(self) -> str:
pass
@property
@abc.abstractmethod
def event_time(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def data_version(self) -> str:
pass
class Document(abc.ABC):
@classmethod
@abc.abstractmethod
def from_json(cls, json_data: str) -> 'Document':
pass
@classmethod
@abc.abstractmethod
def from_dict(cls, dct: dict) -> 'Document':
pass
@abc.abstractmethod
def __getitem__(self, key):
pass
@abc.abstractmethod
def __setitem__(self, key, value):
pass
@abc.abstractmethod
def to_json(self) -> str:
pass
@abc.abstractmethod
def to_dict(self) -> dict:
pass
class DocumentList(abc.ABC):
pass
class EventHubEvent(abc.ABC):
@abc.abstractmethod
def get_body(self) -> bytes:
pass
@property
@abc.abstractmethod
def partition_key(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def sequence_number(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def iothub_metadata(self) -> typing.Optional[typing.Mapping[str, str]]:
pass
@property
@abc.abstractmethod
def enqueued_time(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def offset(self) -> typing.Optional[str]:
pass
class OrchestrationContext(abc.ABC):
@property
@abc.abstractmethod
def body(self) -> str:
pass
class ServiceBusMessage(abc.ABC):
@abc.abstractmethod
def get_body(self) -> typing.Union[str, bytes]:
pass
@property
@abc.abstractmethod
def application_properties(self) -> typing.Dict[str, typing.Any]:
pass
@property
@abc.abstractmethod
def content_type(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def correlation_id(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def dead_letter_error_description(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def dead_letter_reason(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def dead_letter_source(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def delivery_count(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def enqueued_sequence_number(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def enqueued_time_utc(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def expires_at_utc(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def expiration_time(self) -> typing.Optional[datetime.datetime]:
"""(Deprecated, use expires_at_utc instead)"""
pass
@property
@abc.abstractmethod
def label(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def locked_until(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def lock_token(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def message_id(self) -> str:
pass
@property
@abc.abstractmethod
def partition_key(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def reply_to(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def reply_to_session_id(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def scheduled_enqueue_time(self) -> typing.Optional[datetime.datetime]:
"""(Deprecated, use scheduled_enqueue_time_utc instead)"""
pass
@property
@abc.abstractmethod
def scheduled_enqueue_time_utc(self) -> typing.Optional[datetime.datetime]:
pass
@property
@abc.abstractmethod
def sequence_number(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def session_id(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def state(self) -> typing.Optional[int]:
pass
@property
@abc.abstractmethod
def subject(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def time_to_live(self) -> typing.Optional[datetime.timedelta]:
pass
@property
@abc.abstractmethod
def to(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def transaction_partition_key(self) -> typing.Optional[str]:
pass
@property
@abc.abstractmethod
def user_properties(self) -> typing.Dict[str, typing.Any]:
pass
@property
@abc.abstractmethod
def metadata(self) -> typing.Optional[typing.Dict[str, typing.Any]]:
pass