Best Python code snippet using Testify_python
context.py
Source:context.py
...114 except AirflowNotFoundException:115 return default_conn116class AirflowContextDeprecationWarning(DeprecationWarning):117 """Warn for usage of deprecated context variables in a task."""118def _create_deprecation_warning(key: str, replacements: List[str]) -> DeprecationWarning:119 message = f"Accessing {key!r} from the template is deprecated and will be removed in a future version."120 if not replacements:121 return AirflowContextDeprecationWarning(message)122 display_except_last = ", ".join(repr(r) for r in replacements[:-1])123 if display_except_last:124 message += f" Please use {display_except_last} or {replacements[-1]!r} instead."125 else:126 message += f" Please use {replacements[-1]!r} instead."127 return AirflowContextDeprecationWarning(message)128class Context(MutableMapping[str, Any]):129 """Jinja2 template context for task rendering.130 This is a mapping (dict-like) class that can lazily emit warnings when131 (and only when) deprecated context keys are accessed.132 """133 _DEPRECATION_REPLACEMENTS: Dict[str, List[str]] = {134 "execution_date": ["data_interval_start", "logical_date"],135 "next_ds": ["{{ data_interval_end | ds }}"],136 "next_ds_nodash": ["{{ data_interval_end | ds_nodash }}"],137 "next_execution_date": ["data_interval_end"],138 "prev_ds": [],139 "prev_ds_nodash": [],140 "prev_execution_date": [],141 "prev_execution_date_success": ["prev_data_interval_start_success"],142 "tomorrow_ds": [],143 "tomorrow_ds_nodash": [],144 "yesterday_ds": [],145 "yesterday_ds_nodash": [],146 }147 def __init__(self, context: Optional[MutableMapping[str, Any]] = None, **kwargs: Any) -> None:148 self._context = context or {}149 if kwargs:150 self._context.update(kwargs)151 self._deprecation_replacements = self._DEPRECATION_REPLACEMENTS.copy()152 def __repr__(self) -> str:153 return repr(self._context)154 def __reduce_ex__(self, protocol: int) -> Tuple[Any, ...]:155 """Pickle the context as a dict.156 We are intentionally going through ``__getitem__`` in this function,157 instead of using ``items()``, to trigger deprecation warnings.158 """159 items = [(key, self[key]) for key in self._context]160 return dict, (items,)161 def __copy__(self) -> "Context":162 new = type(self)(copy.copy(self._context))163 new._deprecation_replacements = self._deprecation_replacements.copy()164 return new165 def __getitem__(self, key: str) -> Any:166 with contextlib.suppress(KeyError):167 warnings.warn(_create_deprecation_warning(key, self._deprecation_replacements[key]))168 with contextlib.suppress(KeyError):169 return self._context[key]170 raise KeyError(key)171 def __setitem__(self, key: str, value: Any) -> None:172 self._deprecation_replacements.pop(key, None)173 self._context[key] = value174 def __delitem__(self, key: str) -> None:175 self._deprecation_replacements.pop(key, None)176 del self._context[key]177 def __contains__(self, key: object) -> bool:178 return key in self._context179 def __iter__(self) -> Iterator[str]:180 return iter(self._context)181 def __len__(self) -> int:182 return len(self._context)183 def __eq__(self, other: Any) -> bool:184 if not isinstance(other, Context):185 return NotImplemented186 return self._context == other._context187 def __ne__(self, other: Any) -> bool:188 if not isinstance(other, Context):189 return NotImplemented190 return self._context != other._context191 def keys(self) -> AbstractSet[str]:192 return self._context.keys()193 def items(self):194 return ItemsView(self._context)195 def values(self):196 return ValuesView(self._context)197def context_merge(context: "Context", *args: Any, **kwargs: Any) -> None:198 """Merge parameters into an existing context.199 Like ``dict.update()`` , this take the same parameters, and updates200 ``context`` in-place.201 This is implemented as a free function because the ``Context`` type is202 "faked" as a ``TypedDict`` in ``context.pyi``, which cannot have custom203 functions.204 :meta private:205 """206 context.update(*args, **kwargs)207def context_copy_partial(source: "Context", keys: Container[str]) -> "Context":208 """Create a context by copying items under selected keys in ``source``.209 This is implemented as a free function because the ``Context`` type is210 "faked" as a ``TypedDict`` in ``context.pyi``, which cannot have custom211 functions.212 :meta private:213 """214 new = Context({k: v for k, v in source._context.items() if k in keys})215 new._deprecation_replacements = source._deprecation_replacements.copy()216 return new217def lazy_mapping_from_context(source: Context) -> Mapping[str, Any]:218 """Create a mapping that wraps deprecated entries in a lazy object proxy.219 This further delays deprecation warning to until when the entry is actually220 used, instead of when it's accessed in the context. The result is useful for221 passing into a callable with ``**kwargs``, which would unpack the mapping222 too eagerly otherwise.223 This is implemented as a free function because the ``Context`` type is224 "faked" as a ``TypedDict`` in ``context.pyi``, which cannot have custom225 functions.226 :meta private:227 """228 def _deprecated_proxy_factory(k: str, v: Any) -> Any:229 replacements = source._deprecation_replacements[k]230 warnings.warn(_create_deprecation_warning(k, replacements))231 return v232 def _create_value(k: str, v: Any) -> Any:233 if k not in source._deprecation_replacements:234 return v235 factory = functools.partial(_deprecated_proxy_factory, k, v)236 return lazy_object_proxy.Proxy(factory)...
Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!