flytekit.core.shim_task
Directory
Classes
Class | Description |
---|---|
Any |
Special type indicating an unconstrained type. |
ExecutableTemplateShimTask |
The canonical @task decorated Python function task is pretty simple to reason about. |
ExecutionParameters |
This is a run-time user-centric context object that is accessible to every @task method. |
ExecutionState |
This is the context that is active when executing a task or a local workflow. |
FlyteContext |
This is an internal-facing context object, that most users will not have to deal with. |
FlyteContextManager |
FlyteContextManager manages the execution context within Flytekit. |
Generic |
Abstract base class for generic types. |
ShimTaskExecutor |
Please see the notes for the metaclass above first. |
TrackedInstance |
Please see the notes for the metaclass above first. |
TypeEngine |
Core Extensible TypeEngine of Flytekit. |
TypeVar |
Type variable. |
flytekit.core.shim_task.Any
Special type indicating an unconstrained type.
- Any is compatible with every type.
- Any assumed to have all methods.
- All values assumed to be instances of Any.
Note that all the above statements are true from the point of view of static type checkers. At runtime, Any should not be used with instance checks.
flytekit.core.shim_task.ExecutableTemplateShimTask
The canonical @task
decorated Python function task is pretty simple to reason about. At execution time (either
locally or on a Flyte cluster), the function runs.
This class, along with the ShimTaskExecutor
class below, represents another execution pattern. This pattern,
has two components:
- The
TaskTemplate
, or something like it like aFlyteTask
. - An executor, which can use information from the task template (including the
custom
field)
Basically at execution time (both locally and on a Flyte cluster), the task template is given to the executor, which is responsible for computing and returning the results.
.. note::
The interface at execution time will have to derived from the Flyte IDL interface, which means it may be lossy.
This is because when a task is serialized from Python into the TaskTemplate
some information is lost because
Flyte IDL can’t keep track of every single Python type (or Java type if writing in the Java flytekit).
This class also implements the dispatch_execute
and execute
functions to make it look like a PythonTask
that the entrypoint.py
can execute, even though this class doesn’t inherit from PythonTask
.
def ExecutableTemplateShimTask(
tt: _task_model.TaskTemplate,
executor_type: Type[ShimTaskExecutor],
args,
kwargs,
):
Parameter | Type |
---|---|
tt |
_task_model.TaskTemplate |
executor_type |
Type[ShimTaskExecutor] |
args |
*args |
kwargs |
**kwargs |
Methods
Method | Description |
---|---|
dispatch_execute() |
This function is largely similar to the base PythonTask, with the exception that we have to infer the Python |
execute() |
Rather than running here, send everything to the executor |
post_execute() |
This function is a stub, just here to keep dispatch_execute compatibility between this class and PythonTask |
pre_execute() |
This function is a stub, just here to keep dispatch_execute compatibility between this class and PythonTask |
dispatch_execute()
def dispatch_execute(
ctx: FlyteContext,
input_literal_map: _literal_models.LiteralMap,
):
This function is largely similar to the base PythonTask, with the exception that we have to infer the Python
interface before executing. Also, we refer to self.task_template
rather than just self
similar to task
classes that derive from the base PythonTask
.
Parameter | Type |
---|---|
ctx |
FlyteContext |
input_literal_map |
_literal_models.LiteralMap |
execute()
def execute(
kwargs,
):
Rather than running here, send everything to the executor.
Parameter | Type |
---|---|
kwargs |
**kwargs |
post_execute()
def post_execute(
_: Optional[ExecutionParameters],
rval: Any,
):
This function is a stub, just here to keep dispatch_execute compatibility between this class and PythonTask.
Parameter | Type |
---|---|
_ |
Optional[ExecutionParameters] |
rval |
Any |
pre_execute()
def pre_execute(
user_params: Optional[ExecutionParameters],
):
This function is a stub, just here to keep dispatch_execute compatibility between this class and PythonTask.
Parameter | Type |
---|---|
user_params |
Optional[ExecutionParameters] |
Properties
Property | Type | Description |
---|---|---|
executor | ||
executor_type | ||
name | ||
task_template |
flytekit.core.shim_task.ExecutionParameters
This is a run-time user-centric context object that is accessible to every @task method. It can be accessed using
.. code-block:: python
flytekit.current_context()
This object provides the following
- a statsd handler
- a logging handler
- the execution ID as an :py:class:
flytekit.models.core.identifier.WorkflowExecutionIdentifier
object - a working directory for the user to write arbitrary files to
Please do not confuse this object with the :py:class:flytekit.FlyteContext
object.
def ExecutionParameters(
execution_date,
tmp_dir,
stats,
execution_id: typing.Optional[_identifier.WorkflowExecutionIdentifier],
logging,
raw_output_prefix,
output_metadata_prefix,
checkpoint,
decks,
task_id: typing.Optional[_identifier.Identifier],
enable_deck: bool,
kwargs,
):
Parameter | Type |
---|---|
execution_date |
|
tmp_dir |
|
stats |
|
execution_id |
typing.Optional[_identifier.WorkflowExecutionIdentifier] |
logging |
|
raw_output_prefix |
|
output_metadata_prefix |
|
checkpoint |
|
decks |
|
task_id |
typing.Optional[_identifier.Identifier] |
enable_deck |
bool |
kwargs |
**kwargs |
Methods
Method | Description |
---|---|
builder() |
None |
get() |
Returns task specific context if present else raise an error |
has_attr() |
None |
new_builder() |
None |
with_enable_deck() |
None |
with_task_sandbox() |
None |
builder()
def builder()
get()
def get(
key: str,
):
Returns task specific context if present else raise an error. The returned context will match the key
Parameter | Type |
---|---|
key |
str |
has_attr()
def has_attr(
attr_name: str,
):
Parameter | Type |
---|---|
attr_name |
str |
new_builder()
def new_builder(
current: Optional[ExecutionParameters],
):
Parameter | Type |
---|---|
current |
Optional[ExecutionParameters] |
with_enable_deck()
def with_enable_deck(
enable_deck: bool,
):
Parameter | Type |
---|---|
enable_deck |
bool |
with_task_sandbox()
def with_task_sandbox()
Properties
Property | Type | Description |
---|---|---|
checkpoint | ||
decks | ||
default_deck | ||
enable_deck | ||
execution_date | ||
execution_id | ||
logging | ||
output_metadata_prefix | ||
raw_output_prefix | ||
secrets | ||
stats | ||
task_id | ||
timeline_deck | ||
working_directory |
flytekit.core.shim_task.ExecutionState
This is the context that is active when executing a task or a local workflow. This carries the necessary state to execute. Some required things during execution deal with temporary directories, ExecutionParameters that are passed to the user etc.
Attributes: mode (ExecutionState.Mode): Defines the context in which the task is executed (local, hosted, etc). working_dir (os.PathLike): Specifies the remote, external directory where inputs, outputs and other protobufs are uploaded engine_dir (os.PathLike): branch_eval_mode Optional[BranchEvalMode]: Used to determine whether a branch node should execute. user_space_params Optional[ExecutionParameters]: Provides run-time, user-centric context such as a statsd handler, a logging handler, the current execution id and a working directory.
def ExecutionState(
working_dir: Union[os.PathLike, str],
mode: Optional[ExecutionState.Mode],
engine_dir: Optional[Union[os.PathLike, str]],
branch_eval_mode: Optional[BranchEvalMode],
user_space_params: Optional[ExecutionParameters],
):
Parameter | Type |
---|---|
working_dir |
Union[os.PathLike, str] |
mode |
Optional[ExecutionState.Mode] |
engine_dir |
Optional[Union[os.PathLike, str]] |
branch_eval_mode |
Optional[BranchEvalMode] |
user_space_params |
Optional[ExecutionParameters] |
Methods
Method | Description |
---|---|
branch_complete() |
Indicates that we are within a conditional / ifelse block and the active branch is not done |
is_local_execution() |
None |
take_branch() |
Indicates that we are within an if-else block and the current branch has evaluated to true |
with_params() |
Produces a copy of the current execution state and overrides the copy’s parameters with passed parameter values |
branch_complete()
def branch_complete()
Indicates that we are within a conditional / ifelse block and the active branch is not done. Default to SKIPPED
is_local_execution()
def is_local_execution()
take_branch()
def take_branch()
Indicates that we are within an if-else block and the current branch has evaluated to true. Useful only in local execution mode
with_params()
def with_params(
working_dir: Optional[os.PathLike],
mode: Optional[Mode],
engine_dir: Optional[os.PathLike],
branch_eval_mode: Optional[BranchEvalMode],
user_space_params: Optional[ExecutionParameters],
):
Produces a copy of the current execution state and overrides the copy’s parameters with passed parameter values.
Parameter | Type |
---|---|
working_dir |
Optional[os.PathLike] |
mode |
Optional[Mode] |
engine_dir |
Optional[os.PathLike] |
branch_eval_mode |
Optional[BranchEvalMode] |
user_space_params |
Optional[ExecutionParameters] |
flytekit.core.shim_task.FlyteContext
This is an internal-facing context object, that most users will not have to deal with. It’s essentially a globally available grab bag of settings and objects that allows flytekit to do things like convert complex types, run and compile workflows, serialize Flyte entities, etc.
Even though this object as a current_context
function on it, it should not be called directly. Please use the
:py:class:flytekit.FlyteContextManager
object instead.
Please do not confuse this object with the :py:class:flytekit.ExecutionParameters
object.
def FlyteContext(
file_access: FileAccessProvider,
level: int,
flyte_client: Optional['friendly_client.SynchronousFlyteClient'],
compilation_state: Optional[CompilationState],
execution_state: Optional[ExecutionState],
serialization_settings: Optional[SerializationSettings],
in_a_condition: bool,
origin_stackframe: Optional[traceback.FrameSummary],
output_metadata_tracker: Optional[OutputMetadataTracker],
worker_queue: Optional[Controller],
):
Parameter | Type |
---|---|
file_access |
FileAccessProvider |
level |
int |
flyte_client |
Optional['friendly_client.SynchronousFlyteClient'] |
compilation_state |
Optional[CompilationState] |
execution_state |
Optional[ExecutionState] |
serialization_settings |
Optional[SerializationSettings] |
in_a_condition |
bool |
origin_stackframe |
Optional[traceback.FrameSummary] |
output_metadata_tracker |
Optional[OutputMetadataTracker] |
worker_queue |
Optional[Controller] |
Methods
Method | Description |
---|---|
current_context() |
This method exists only to maintain backwards compatibility |
enter_conditional_section() |
None |
get_deck() |
Returns the deck that was created as part of the last execution |
get_origin_stackframe_repr() |
None |
new_builder() |
None |
new_compilation_state() |
Creates and returns a default compilation state |
new_execution_state() |
Creates and returns a new default execution state |
set_stackframe() |
None |
with_client() |
None |
with_compilation_state() |
None |
with_execution_state() |
None |
with_file_access() |
None |
with_new_compilation_state() |
None |
with_output_metadata_tracker() |
None |
with_serialization_settings() |
None |
with_worker_queue() |
None |
current_context()
def current_context()
This method exists only to maintain backwards compatibility. Please use
FlyteContextManager.current_context()
instead.
Users of flytekit should be wary not to confuse the object returned from this function
with :py:func:flytekit.current_context
enter_conditional_section()
def enter_conditional_section()
get_deck()
def get_deck()
Returns the deck that was created as part of the last execution.
The return value depends on the execution environment. In a notebook, the return value is compatible with IPython.display and should be rendered in the notebook.
.. code-block:: python
with flytekit.new_context() as ctx: my_task(…) ctx.get_deck()
OR if you wish to explicitly display
.. code-block:: python
from IPython import display display(ctx.get_deck())
get_origin_stackframe_repr()
def get_origin_stackframe_repr()
new_builder()
def new_builder()
new_compilation_state()
def new_compilation_state(
prefix: str,
):
Creates and returns a default compilation state. For most of the code this should be the entrypoint of compilation, otherwise the code should always uses - with_compilation_state
Parameter | Type |
---|---|
prefix |
str |
new_execution_state()
def new_execution_state(
working_dir: Optional[os.PathLike],
):
Creates and returns a new default execution state. This should be used at the entrypoint of execution, in all other cases it is preferable to use with_execution_state
Parameter | Type |
---|---|
working_dir |
Optional[os.PathLike] |
set_stackframe()
def set_stackframe(
s: traceback.FrameSummary,
):
Parameter | Type |
---|---|
s |
traceback.FrameSummary |
with_client()
def with_client(
c: SynchronousFlyteClient,
):
Parameter | Type |
---|---|
c |
SynchronousFlyteClient |
with_compilation_state()
def with_compilation_state(
c: CompilationState,
):
Parameter | Type |
---|---|
c |
CompilationState |
with_execution_state()
def with_execution_state(
es: ExecutionState,
):
Parameter | Type |
---|---|
es |
ExecutionState |
with_file_access()
def with_file_access(
fa: FileAccessProvider,
):
Parameter | Type |
---|---|
fa |
FileAccessProvider |
with_new_compilation_state()
def with_new_compilation_state()
with_output_metadata_tracker()
def with_output_metadata_tracker(
t: OutputMetadataTracker,
):
Parameter | Type |
---|---|
t |
OutputMetadataTracker |
with_serialization_settings()
def with_serialization_settings(
ss: SerializationSettings,
):
Parameter | Type |
---|---|
ss |
SerializationSettings |
with_worker_queue()
def with_worker_queue(
wq: Controller,
):
Parameter | Type |
---|---|
wq |
Controller |
Properties
Property | Type | Description |
---|---|---|
user_space_params |
flytekit.core.shim_task.FlyteContextManager
FlyteContextManager manages the execution context within Flytekit. It holds global state of either compilation
or Execution. It is not thread-safe and can only be run as a single threaded application currently.
Context’s within Flytekit is useful to manage compilation state and execution state. Refer to CompilationState
and ExecutionState
for more information. FlyteContextManager provides a singleton stack to manage these contexts.
Typical usage is
.. code-block:: python
FlyteContextManager.initialize() with FlyteContextManager.with_context(o) as ctx: pass
If required - not recommended you can use
FlyteContextManager.push_context()
but correspondingly a pop_context should be called
FlyteContextManager.pop_context()
Methods
Method | Description |
---|---|
add_signal_handler() |
None |
current_context() |
None |
get_origin_stackframe() |
None |
initialize() |
Re-initializes the context and erases the entire context |
pop_context() |
None |
push_context() |
None |
size() |
None |
with_context() |
None |
add_signal_handler()
def add_signal_handler(
handler: typing.Callable[[int, FrameType], typing.Any],
):
Parameter | Type |
---|---|
handler |
typing.Callable[[int, FrameType], typing.Any] |
current_context()
def current_context()
get_origin_stackframe()
def get_origin_stackframe(
limit,
):
Parameter | Type |
---|---|
limit |
initialize()
def initialize()
Re-initializes the context and erases the entire context
pop_context()
def pop_context()
push_context()
def push_context(
ctx: FlyteContext,
f: Optional[traceback.FrameSummary],
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
f |
Optional[traceback.FrameSummary] |
size()
def size()
with_context()
def with_context(
b: FlyteContext.Builder,
):
Parameter | Type |
---|---|
b |
FlyteContext.Builder |
flytekit.core.shim_task.Generic
Abstract base class for generic types.
On Python 3.12 and newer, generic classes implicitly inherit from Generic when they declare a parameter list after the class’s name::
class Mapping[KT, VT]: def getitem(self, key: KT) -> VT: …
Etc.
On older versions of Python, however, generic classes have to explicitly inherit from Generic.
After a class has been declared to be generic, it can then be used as follows::
def lookup_name[KT, VT](mapping: Mapping[KT, VT], key: KT, default: VT) -> VT: try: return mapping[key] except KeyError: return default
flytekit.core.shim_task.ShimTaskExecutor
Please see the notes for the metaclass above first.
This functionality has two use-cases currently,
- Keep track of naming for non-function
PythonAutoContainerTasks
. That is, things like the :py:class:flytekit.extras.sqlite3.task.SQLite3Task
task. - Task resolvers, because task resolvers are instances of :py:class:
flytekit.core.python_auto_container.TaskResolverMixin
classes, not the classes themselves, which means we need to look on the left hand side of them to see how to find them at task execution time.
def ShimTaskExecutor(
args,
kwargs,
):
Parameter | Type |
---|---|
args |
*args |
kwargs |
**kwargs |
Methods
Method | Description |
---|---|
execute_from_model() |
This function must be overridden and is where all the business logic for running a task should live |
find_lhs() |
None |
execute_from_model()
def execute_from_model(
tt: _task_model.TaskTemplate,
kwargs,
):
This function must be overridden and is where all the business logic for running a task should live. Keep in
mind that you’re only working with the TaskTemplate
. You won’t have access to any information in the task
that wasn’t serialized into the template.
Parameter | Type |
---|---|
tt |
_task_model.TaskTemplate |
kwargs |
**kwargs |
find_lhs()
def find_lhs()
Properties
Property | Type | Description |
---|---|---|
instantiated_in | ||
lhs | ||
location |
flytekit.core.shim_task.TrackedInstance
Please see the notes for the metaclass above first.
This functionality has two use-cases currently,
- Keep track of naming for non-function
PythonAutoContainerTasks
. That is, things like the :py:class:flytekit.extras.sqlite3.task.SQLite3Task
task. - Task resolvers, because task resolvers are instances of :py:class:
flytekit.core.python_auto_container.TaskResolverMixin
classes, not the classes themselves, which means we need to look on the left hand side of them to see how to find them at task execution time.
def TrackedInstance(
args,
kwargs,
):
Parameter | Type |
---|---|
args |
*args |
kwargs |
**kwargs |
Methods
Method | Description |
---|---|
find_lhs() |
None |
find_lhs()
def find_lhs()
Properties
Property | Type | Description |
---|---|---|
instantiated_in | ||
lhs | ||
location |
flytekit.core.shim_task.TypeEngine
Core Extensible TypeEngine of Flytekit. This should be used to extend the capabilities of FlyteKits type system. Users can implement their own TypeTransformers and register them with the TypeEngine. This will allow special handling of user objects
Methods
Method | Description |
---|---|
async_to_literal() |
Converts a python value of a given type and expected LiteralType into a resolved Literal value |
async_to_python_value() |
None |
calculate_hash() |
None |
dict_to_literal_map() |
None |
dict_to_literal_map_pb() |
None |
get_available_transformers() |
Returns all python types for which transformers are available |
get_transformer() |
Implements a recursive search for the transformer |
guess_python_type() |
Transforms a flyte-specific LiteralType to a regular python value |
guess_python_types() |
Transforms a dictionary of flyte-specific Variable objects to a dictionary of regular python values |
lazy_import_transformers() |
Only load the transformers if needed |
literal_map_to_kwargs() |
None |
named_tuple_to_variable_map() |
Converts a python-native NamedTuple to a flyte-specific VariableMap of named literals |
register() |
This should be used for all types that respond with the right type annotation when you use type( |
register_additional_type() |
None |
register_restricted_type() |
None |
to_html() |
None |
to_literal() |
The current dance is because we are allowing users to call from an async function, this synchronous |
to_literal_checks() |
None |
to_literal_type() |
Converts a python type into a flyte specific LiteralType |
to_python_value() |
Converts a Literal value with an expected python type into a python value |
unwrap_offloaded_literal() |
None |
async_to_literal()
def async_to_literal(
ctx: FlyteContext,
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
Converts a python value of a given type and expected LiteralType
into a resolved Literal
value.
Parameter | Type |
---|---|
ctx |
FlyteContext |
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
async_to_python_value()
def async_to_python_value(
ctx: FlyteContext,
lv: Literal,
expected_python_type: Type,
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
lv |
Literal |
expected_python_type |
Type |
calculate_hash()
def calculate_hash(
python_val: typing.Any,
python_type: Type[T],
):
Parameter | Type |
---|---|
python_val |
typing.Any |
python_type |
Type[T] |
dict_to_literal_map()
def dict_to_literal_map(
ctx: FlyteContext,
d: typing.Dict[str, typing.Any],
type_hints: Optional[typing.Dict[str, type]],
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
d |
typing.Dict[str, typing.Any] |
type_hints |
Optional[typing.Dict[str, type]] |
dict_to_literal_map_pb()
def dict_to_literal_map_pb(
ctx: FlyteContext,
d: typing.Dict[str, typing.Any],
type_hints: Optional[typing.Dict[str, type]],
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
d |
typing.Dict[str, typing.Any] |
type_hints |
Optional[typing.Dict[str, type]] |
get_available_transformers()
def get_available_transformers()
Returns all python types for which transformers are available
get_transformer()
def get_transformer(
python_type: Type,
):
Implements a recursive search for the transformer.
Parameter | Type |
---|---|
python_type |
Type |
guess_python_type()
def guess_python_type(
flyte_type: LiteralType,
):
Transforms a flyte-specific LiteralType
to a regular python value.
Parameter | Type |
---|---|
flyte_type |
LiteralType |
guess_python_types()
def guess_python_types(
flyte_variable_dict: typing.Dict[str, _interface_models.Variable],
):
Transforms a dictionary of flyte-specific Variable
objects to a dictionary of regular python values.
Parameter | Type |
---|---|
flyte_variable_dict |
typing.Dict[str, _interface_models.Variable] |
lazy_import_transformers()
def lazy_import_transformers()
Only load the transformers if needed.
literal_map_to_kwargs()
def literal_map_to_kwargs(
ctx: FlyteContext,
lm: LiteralMap,
python_types: typing.Optional[typing.Dict[str, type]],
literal_types: typing.Optional[typing.Dict[str, _interface_models.Variable]],
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
lm |
LiteralMap |
python_types |
typing.Optional[typing.Dict[str, type]] |
literal_types |
typing.Optional[typing.Dict[str, _interface_models.Variable]] |
named_tuple_to_variable_map()
def named_tuple_to_variable_map(
t: typing.NamedTuple,
):
Converts a python-native NamedTuple
to a flyte-specific VariableMap of named literals.
Parameter | Type |
---|---|
t |
typing.NamedTuple |
register()
def register(
transformer: TypeTransformer,
additional_types: Optional[typing.List[Type]],
):
This should be used for all types that respond with the right type annotation when you use type(…) function
Parameter | Type |
---|---|
transformer |
TypeTransformer |
additional_types |
Optional[typing.List[Type]] |
register_additional_type()
def register_additional_type(
transformer: TypeTransformer[T],
additional_type: Type[T],
override,
):
Parameter | Type |
---|---|
transformer |
TypeTransformer[T] |
additional_type |
Type[T] |
override |
register_restricted_type()
def register_restricted_type(
name: str,
type: Type[T],
):
Parameter | Type |
---|---|
name |
str |
type |
Type[T] |
to_html()
def to_html(
ctx: FlyteContext,
python_val: typing.Any,
expected_python_type: Type[typing.Any],
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
python_val |
typing.Any |
expected_python_type |
Type[typing.Any] |
to_literal()
def to_literal(
ctx: FlyteContext,
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
The current dance is because we are allowing users to call from an async function, this synchronous to_literal function, and allowing this to_literal function, to then invoke yet another async function, namely an async transformer.
Parameter | Type |
---|---|
ctx |
FlyteContext |
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
to_literal_checks()
def to_literal_checks(
python_val: typing.Any,
python_type: Type[T],
expected: LiteralType,
):
Parameter | Type |
---|---|
python_val |
typing.Any |
python_type |
Type[T] |
expected |
LiteralType |
to_literal_type()
def to_literal_type(
python_type: Type[T],
):
Converts a python type into a flyte specific LiteralType
Parameter | Type |
---|---|
python_type |
Type[T] |
to_python_value()
def to_python_value(
ctx: FlyteContext,
lv: Literal,
expected_python_type: Type,
):
Converts a Literal value with an expected python type into a python value.
Parameter | Type |
---|---|
ctx |
FlyteContext |
lv |
Literal |
expected_python_type |
Type |
unwrap_offloaded_literal()
def unwrap_offloaded_literal(
ctx: FlyteContext,
lv: Literal,
):
Parameter | Type |
---|---|
ctx |
FlyteContext |
lv |
Literal |
flytekit.core.shim_task.TypeVar
Type variable.
The preferred way to construct a type variable is via the dedicated syntax for generic functions, classes, and type aliases::
class Sequence[T]: # T is a TypeVar …
This syntax can also be used to create bound and constrained type variables::
S is a TypeVar bound to str
class StrSequence[S: str]: …
A is a TypeVar constrained to str or bytes
class StrOrBytesSequence[A: (str, bytes)]: …
However, if desired, reusable type variables can also be constructed manually, like so::
T = TypeVar(‘T’) # Can be anything S = TypeVar(‘S’, bound=str) # Can be any subtype of str A = TypeVar(‘A’, str, bytes) # Must be exactly str or bytes
Type variables exist primarily for the benefit of static type checkers. They serve as the parameters for generic types as well as for generic function and type alias definitions.
The variance of type variables is inferred by type checkers when they
are created through the type parameter syntax and when
infer_variance=True
is passed. Manually created type variables may
be explicitly marked covariant or contravariant by passing
covariant=True
or contravariant=True
. By default, manually
created type variables are invariant. See PEP 484 and PEP 695 for more
details.