lint fix

dlt/destinations/impl/bigquery/bigquery.py

@@ -1,7 +1,7 @@
 import functools
 import os
 from pathlib import Path
-from typing import ClassVar, Optional, Sequence, Tuple, List, cast, Dict
+from typing import Any, ClassVar, Dict, List, Optional, Sequence, Tuple, Type, cast
 
 import google.cloud.bigquery as bigquery  # noqa: I250
 from google.api_core import exceptions as api_core_exceptions
@@ -225,9 +225,9 @@ def start_file_load(self, table: TTableSchema, file_path: str, load_id: str) ->
         insert_api = table.get("insert_api", self.config.loading_api)
         if insert_api == "streaming":
             if file_path.endswith(".jsonl"):
-                job_cls = DestinationJsonlLoadJob
+                job_cls: Type[DestinationJsonlLoadJob] = DestinationJsonlLoadJob
             elif file_path.endswith(".parquet"):
-                job_cls = DestinationParquetLoadJob
+                job_cls: Type[DestinationParquetLoadJob] = DestinationParquetLoadJob
             else:
                 raise ValueError(
                     f"Unsupported file type for BigQuery streaming inserts: {file_path}"
@@ -236,18 +236,18 @@ def start_file_load(self, table: TTableSchema, file_path: str, load_id: str) ->
             job = job_cls(
                 table,
                 file_path,
-                self.config,
+                self.config,  # type: ignore
                 self.schema,
                 destination_state(),
                 functools.partial(_streaming_load, self.sql_client),
                 [],
             )
         else:
-            job = super().start_file_load(table, file_path, load_id)
+            job = super().start_file_load(table, file_path, load_id)  # type: ignore
 
             if not job:
                 try:
-                    job = BigQueryLoadJob(
+                    job = BigQueryLoadJob(  # type: ignore
                         FileStorage.get_file_name_from_file_path(file_path),
                         self._create_load_job(table, file_path),
                         self.config.http_timeout,
@@ -451,7 +451,7 @@ def _from_db_type(
         return self.type_mapper.from_db_type(bq_t, precision, scale)
 
 
-def _streaming_load(sql_client, items, table):
+def _streaming_load(sql_client: BigQueryClient, items: List[Dict[Any, Any]], table: Dict[Any, Any]):
     """
     Upload the given items into BigQuery table, using streaming API.
     Streaming API is used for small amounts of data, with optimal
@@ -464,7 +464,7 @@ def _streaming_load(sql_client, items, table):
         table (Dict[Any, Any]): Table schema.
     """
 
-    def _should_retry(exc):
+    def _should_retry(exc: Exception) -> bool:
         """Predicate to decide if we need to retry the exception.
 
         Args:

dlt/destinations/impl/bigquery/bigquery_adapter.py

@@ -30,7 +30,7 @@ def bigquery_adapter(
     round_half_even: TColumnNames = None,
     table_description: Optional[str] = None,
     table_expiration_datetime: Optional[str] = None,
-    insert_api: Optional[Literal["streaming_insert", "default"]] = "default",
+    insert_api: Optional[Literal["streaming", "default"]] = "default",
 ) -> DltResource:
     """
     Prepares data for loading into BigQuery.

dlt/destinations/impl/bigquery/sql_client.py

@@ -48,7 +48,7 @@ class BigQueryDBApiCursorImpl(DBApiCursorImpl):
     def df(self, chunk_size: int = None, **kwargs: Any) -> DataFrame:
         if chunk_size is not None:
             return super().df(chunk_size=chunk_size)
-        query_job: bigquery.QueryJob = self.native_cursor._query_job
+        query_job: bigquery.QueryJob = self.native_cursor._query_job  # type: ignore
 
         try:
             return query_job.to_dataframe(**kwargs)