temporalio.worker.workflow_sandbox._importer.Importer

class documentation

class Importer: (source)

Constructor: Importer(restrictions, restriction_context)

View In Hierarchy

Importer that restricts modules.

Static Method current_importer Get the current importer if any.
Method __init__ Create importer.
Method applied Context manager to apply this restrictive import.
Method module_configured_passthrough Whether the given module name is configured as passthrough.
Instance Variable import_func Undocumented
Instance Variable modules_checked_for_restrictions Undocumented
Instance Variable new_modules Undocumented
Instance Variable restricted_builtins Undocumented
Instance Variable restriction_context Undocumented
Instance Variable restrictions Undocumented
Method _assert_valid_module Undocumented
Method _builtins_restricted Undocumented
Method _builtins_unrestricted Undocumented
Method _import Undocumented
Method _is_import_notification_policy_applied Undocumented
Method _maybe_passthrough_module Undocumented
Method _maybe_restrict_module Implements _Environment.maybe_restrict_module.
Method _traced_import Undocumented
Method _unapplied Undocumented
Class Variable _thread_local_current Undocumented
@staticmethod
def current_importer() -> Importer | None: (source)

Get the current importer if any.

def __init__(self, restrictions: SandboxRestrictions, restriction_context: RestrictionContext): (source)

Create importer.

@contextmanager
def applied(self) -> Iterator[None]: (source)

Context manager to apply this restrictive import.

Warning

This currently alters global sys.modules and builtins.__import__ while it is running and therefore should be locked against other code running at the same time.

def module_configured_passthrough(self, name: str) -> bool: (source)

Whether the given module name is configured as passthrough.

import_func = (source)

Undocumented

modules_checked_for_restrictions: set[str] = (source)

Undocumented

new_modules: dict[str, types.ModuleType] = (source)

Undocumented

restricted_builtins: list[tuple[str, _ThreadLocalCallable, Callable]] = (source)

Undocumented

restriction_context = (source)

Undocumented

restrictions = (source)

Undocumented

def _assert_valid_module(self, name: str): (source)

Undocumented

@contextmanager
def _builtins_restricted(self) -> Iterator[None]: (source)

Undocumented

@contextmanager
def _builtins_unrestricted(self) -> Iterator[None]: (source)

Undocumented

def _import(self, name: str, globals: Mapping[str, object] | None = None, locals: Mapping[str, object] | None = None, fromlist: Sequence[str] = (), level: int = 0) -> types.ModuleType: (source)

Undocumented

def _is_import_notification_policy_applied(self, policy: temporalio.workflow.SandboxImportNotificationPolicy) -> bool: (source)

Undocumented

def _maybe_passthrough_module(self, name: str) -> types.ModuleType | None: (source)

Undocumented

def _maybe_restrict_module(self, mod: types.ModuleType) -> types.ModuleType | None: (source)

Implements _Environment.maybe_restrict_module.

def _traced_import(self, name: str, globals: Mapping[str, object] | None = None, locals: Mapping[str, object] | None = None, fromlist: Sequence[str] = (), level: int = 0) -> types.ModuleType: (source)

Undocumented

@contextmanager
def _unapplied(self) -> Iterator[None]: (source)

Undocumented

_thread_local_current = (source)

Undocumented