Fix mypy providers

airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py

@@ -35,7 +35,7 @@
     catchup=False,
     tags=['example'],
 ) as dag:
-    table_sensor = CassandraTableSensor(task_id="cassandra_table_sensor")
+    table_sensor = CassandraTableSensor(task_id="cassandra_table_sensor", table="test")
 
-    record_sensor = CassandraRecordSensor(task_id="cassandra_record_sensor", keys={"p1": "v1", "p2": "v2"})
+    record_sensor = CassandraRecordSensor(task_id="cassandra_record_sensor", table="test", keys={"p1": "v1", "p2": "v2"})
 # [END howto_operator_cassandra_sensors]

airflow/providers/apache/spark/hooks/spark_sql.py

@@ -17,7 +17,7 @@
 # under the License.
 #
 import subprocess
-from typing import TYPE_CHECKING, Any, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 from airflow.exceptions import AirflowException, AirflowNotFoundException
 from airflow.hooks.base import BaseHook
@@ -83,12 +83,13 @@ def __init__(
         yarn_queue: Optional[str] = None,
     ) -> None:
         super().__init__()
+        options: Dict = {}
+        conn: Optional[Connection] = None
 
         try:
-            conn: "Optional[Connection]" = self.get_connection(conn_id)
+            conn = self.get_connection(conn_id)
         except AirflowNotFoundException:
             conn = None
-            options = {}
         else:
             options = conn.extra_dejson
 

airflow/providers/asana/example_dags/example_asana.py

@@ -28,12 +28,12 @@
     AsanaUpdateTaskOperator,
 )
 
-ASANA_TASK_TO_UPDATE = os.environ.get("ASANA_TASK_TO_UPDATE")
-ASANA_TASK_TO_DELETE = os.environ.get("ASANA_TASK_TO_DELETE")
+ASANA_TASK_TO_UPDATE = os.getenv("ASANA_TASK_TO_UPDATE", "update_task")
+ASANA_TASK_TO_DELETE = os.getenv("ASANA_TASK_TO_DELETE", "delete_task")
 # This example assumes a default project ID has been specified in the connection. If you
 # provide a different id in ASANA_PROJECT_ID_OVERRIDE, it will override this default
 # project ID in the AsanaFindTaskOperator example below
-ASANA_PROJECT_ID_OVERRIDE = os.environ.get("ASANA_PROJECT_ID_OVERRIDE")
+ASANA_PROJECT_ID_OVERRIDE = os.getenv("ASANA_PROJECT_ID_OVERRIDE", "test_project")
 # This connection should specify a personal access token and a default project ID
 CONN_ID = os.environ.get("ASANA_CONNECTION_ID")
 
@@ -53,6 +53,7 @@
     create = AsanaCreateTaskOperator(
         task_id="run_asana_create_task",
         task_parameters={"notes": "Some notes about the task."},
+        conn_id='asana_default',
         name="New Task Name",
     )
     # [END run_asana_create_task_operator]
@@ -66,6 +67,7 @@
     one_week_ago = (datetime.now() - timedelta(days=7)).strftime("%Y-%m-%d")
     find = AsanaFindTaskOperator(
         task_id="run_asana_find_task",
+        conn_id='asana_default',
         search_parameters={"project": ASANA_PROJECT_ID_OVERRIDE, "modified_since": one_week_ago},
     )
     # [END run_asana_find_task_operator]
@@ -75,6 +77,7 @@
     # task attributes you want to update.
     update = AsanaUpdateTaskOperator(
         task_id="run_asana_update_task",
+        conn_id='asana_default',
         asana_task_gid=ASANA_TASK_TO_UPDATE,
         task_parameters={"notes": "This task was updated!", "completed": True},
     )
@@ -84,6 +87,7 @@
     # Delete a task. This task will complete successfully even if `asana_task_gid` does not exist.
     delete = AsanaDeleteTaskOperator(
         task_id="run_asana_delete_task",
+        conn_id='asana_default',
         asana_task_gid=ASANA_TASK_TO_DELETE,
     )
     # [END run_asana_delete_task_operator]

airflow/providers/asana/hooks/asana.py

@@ -23,7 +23,7 @@
 from asana.error import NotFoundError
 
 try:
-    from functools import cached_property
+    from functools import cached_property  # type: ignore
 except ImportError:
     from cached_property import cached_property
 

airflow/providers/jira/hooks/jira.py

@@ -42,7 +42,7 @@ def __init__(self, jira_conn_id: str = default_conn_name, proxies: Optional[Any]
         super().__init__()
         self.jira_conn_id = jira_conn_id
         self.proxies = proxies
-        self.client = None
+        self.client: Optional[JIRA] = None
         self.get_conn()
 
     def get_conn(self) -> JIRA:

airflow/providers/telegram/hooks/telegram.py

@@ -79,7 +79,7 @@ def get_conn(self) -> telegram.bot.Bot:
         """
         return telegram.bot.Bot(token=self.token)
 
-    def __get_token(self, token: Optional[str], telegram_conn_id: str) -> str:
+    def __get_token(self, token: Optional[str], telegram_conn_id: Optional[str]) -> str:
         """
         Returns the telegram API token
 
@@ -103,7 +103,7 @@ def __get_token(self, token: Optional[str], telegram_conn_id: str) -> str:
 
         raise AirflowException("Cannot get token: No valid Telegram connection supplied.")
 
-    def __get_chat_id(self, chat_id: Optional[str], telegram_conn_id: str) -> Optional[str]:
+    def __get_chat_id(self, chat_id: Optional[str], telegram_conn_id: Optional[str]) -> Optional[str]:
         """
         Returns the telegram chat ID for a chat/channel/group
 

airflow/providers/telegram/operators/telegram.py

@@ -16,7 +16,7 @@
 # specific language governing permissions and limitations
 # under the License.
 """Operator for Telegram"""
-from typing import Optional
+from typing import Dict, Optional
 
 from airflow.exceptions import AirflowException
 from airflow.models import BaseOperator
@@ -70,7 +70,7 @@ def __init__(
 
         super().__init__(**kwargs)
 
-    def execute(self, **kwargs) -> None:
+    def execute(self, context: Dict) -> None:
         """Calls the TelegramHook to post the provided Telegram message"""
         if self.text:
             self.telegram_kwargs['text'] = self.text

airflow/providers/trino/hooks/trino.py

@@ -143,6 +143,7 @@ def run(
         hql,
         autocommit: bool = False,
         parameters: Optional[dict] = None,
+        handler=None
     ) -> None:
         """Execute the statement against Trino. Can be used to create views."""
         return super().run(sql=self._strip_sql(hql), parameters=parameters)