-
Notifications
You must be signed in to change notification settings - Fork 13.7k
/
context.pyi
95 lines (85 loc) · 3.28 KB
/
context.pyi
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
#
# 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.
# This stub exists to "fake" the Context class as a TypedDict to provide
# better typehint and editor support.
#
# Unfortunately 'conn', 'macros', 'var.json', and 'var.value' need to be
# annotated as Any and loose discoverability because we don't know what
# attributes are injected at runtime, and giving them a class would trigger
# undefined attribute errors from Mypy. Hopefully there will be a mechanism to
# declare "these are defined, but don't error if others are accessed" someday.
from typing import Any, Container, Mapping, Optional, Union
from pendulum import DateTime
from airflow.configuration import AirflowConfigParser
from airflow.models.baseoperator import BaseOperator
from airflow.models.dag import DAG
from airflow.models.dagrun import DagRun
from airflow.models.param import ParamsDict
from airflow.models.taskinstance import TaskInstance
from airflow.typing_compat import TypedDict
class _VariableAccessors(TypedDict):
json: Any
value: Any
class VariableAccessor:
def __init__(self, *, deserialize_json: bool) -> None: ...
def get(self, key, default: Any = ...) -> Any: ...
class ConnectionAccessor:
def get(self, key: str, default_conn: Any = None) -> Any: ...
class Context(TypedDict, total=False):
conf: AirflowConfigParser
conn: Any
dag: DAG
dag_run: DagRun
data_interval_end: DateTime
data_interval_start: DateTime
ds: str
ds_nodash: str
execution_date: DateTime
exception: Union[Exception, str, None]
inlets: list
logical_date: DateTime
macros: Any
next_ds: Optional[str]
next_ds_nodash: Optional[str]
next_execution_date: Optional[DateTime]
outlets: list
params: ParamsDict
prev_data_interval_start_success: Optional[DateTime]
prev_data_interval_end_success: Optional[DateTime]
prev_ds: Optional[str]
prev_ds_nodash: Optional[str]
prev_execution_date: Optional[DateTime]
prev_execution_date_success: Optional[DateTime]
prev_start_date_success: Optional[DateTime]
run_id: str
task: BaseOperator
task_instance: TaskInstance
task_instance_key_str: str
test_mode: bool
ti: TaskInstance
tomorrow_ds: str
tomorrow_ds_nodash: str
ts: str
ts_nodash: str
ts_nodash_with_tz: str
var: _VariableAccessors
yesterday_ds: str
yesterday_ds_nodash: str
class AirflowContextDeprecationWarning(DeprecationWarning): ...
def context_copy_partial(source: Context, keys: Container[str]) -> Context: ...
def lazy_mapping_from_context(source: Context) -> Mapping[str, Any]: ...