diff --git a/.gitignore b/.gitignore
index b9daa52f1..8e08cebce 100644
--- a/.gitignore
+++ b/.gitignore
@@ -29,6 +29,7 @@ pip-log.txt
 .nox
 .cache
 .pytest_cache
+.pytype
 
 
 # Mac
diff --git a/google/cloud/bigquery/_helpers.py b/google/cloud/bigquery/_helpers.py
index 100136108..880973843 100644
--- a/google/cloud/bigquery/_helpers.py
+++ b/google/cloud/bigquery/_helpers.py
@@ -20,12 +20,12 @@
 import re
 import six
 
-from google.cloud._helpers import UTC
-from google.cloud._helpers import _date_from_iso8601_date
-from google.cloud._helpers import _datetime_from_microseconds
-from google.cloud._helpers import _RFC3339_MICROS
-from google.cloud._helpers import _RFC3339_NO_FRACTION
-from google.cloud._helpers import _to_bytes
+from google.cloud._helpers import UTC  # type: ignore
+from google.cloud._helpers import _date_from_iso8601_date  # type: ignore
+from google.cloud._helpers import _datetime_from_microseconds  # type: ignore
+from google.cloud._helpers import _RFC3339_MICROS  # type: ignore
+from google.cloud._helpers import _RFC3339_NO_FRACTION  # type: ignore
+from google.cloud._helpers import _to_bytes  # type: ignore
 
 _RFC3339_MICROS_NO_ZULU = "%Y-%m-%dT%H:%M:%S.%f"
 _TIMEONLY_WO_MICROS = "%H:%M:%S"
diff --git a/google/cloud/bigquery/_http.py b/google/cloud/bigquery/_http.py
index 8ee633e64..5dacf8108 100644
--- a/google/cloud/bigquery/_http.py
+++ b/google/cloud/bigquery/_http.py
@@ -14,7 +14,7 @@
 
 """Create / interact with Google BigQuery connections."""
 
-from google.cloud import _http
+from google.cloud import _http  # type: ignore
 
 from google.cloud.bigquery import __version__
 
diff --git a/google/cloud/bigquery/_pandas_helpers.py b/google/cloud/bigquery/_pandas_helpers.py
index 7774ce26b..e5668cd3d 100644
--- a/google/cloud/bigquery/_pandas_helpers.py
+++ b/google/cloud/bigquery/_pandas_helpers.py
@@ -23,13 +23,13 @@
 from six.moves import queue
 
 try:
-    import pandas
+    import pandas  # type: ignore
 except ImportError:  # pragma: NO COVER
     pandas = None
 
 try:
-    import pyarrow
-    import pyarrow.parquet
+    import pyarrow  # type: ignore
+    import pyarrow.parquet  # type: ignore
 except ImportError:  # pragma: NO COVER
     pyarrow = None
 
@@ -350,6 +350,7 @@ def augment_schema(dataframe, current_bq_schema):
     Returns:
         Optional[Sequence[google.cloud.bigquery.schema.SchemaField]]
     """
+    # pytype: disable=attribute-error
     augmented_schema = []
     unknown_type_fields = []
 
@@ -383,6 +384,7 @@ def augment_schema(dataframe, current_bq_schema):
         return None
 
     return augmented_schema
+    # pytype: enable=attribute-error
 
 
 def dataframe_to_arrow(dataframe, bq_schema):
@@ -600,7 +602,7 @@ def _download_table_bqstorage(
 
     # Passing a BQ Storage client in implies that the BigQuery Storage library
     # is available and can be imported.
-    from google.cloud import bigquery_storage
+    from google.cloud import bigquery_storage  # type: ignore
 
     if "$" in table.table_id:
         raise ValueError(
diff --git a/google/cloud/bigquery/_tqdm_helpers.py b/google/cloud/bigquery/_tqdm_helpers.py
index 2fcf2a981..8717b5be4 100644
--- a/google/cloud/bigquery/_tqdm_helpers.py
+++ b/google/cloud/bigquery/_tqdm_helpers.py
@@ -19,7 +19,7 @@
 import warnings
 
 try:
-    import tqdm
+    import tqdm  # type: ignore
 except ImportError:  # pragma: NO COVER
     tqdm = None
 
diff --git a/google/cloud/bigquery/client.py b/google/cloud/bigquery/client.py
index 28cac64ad..52dab037d 100644
--- a/google/cloud/bigquery/client.py
+++ b/google/cloud/bigquery/client.py
@@ -75,6 +75,18 @@
 from google.cloud.bigquery.table import TableReference
 from google.cloud.bigquery.table import RowIterator
 
+# Types needed only for Type Hints
+import datetime
+from google.cloud.bigquery.job import (
+    _AsyncJob,
+    LoadJobConfig,
+    QueryJobConfig,
+    CopyJobConfig,
+    ExtractJobConfig,
+)
+from google.api_core import retry as retries  # type: ignore
+from typing import Any, BinaryIO, Dict, Iterable, Sequence, Tuple, Union
+
 
 _DEFAULT_CHUNKSIZE = 1048576  # 1024 * 1024 B = 1 MB
 _MAX_MULTIPART_SIZE = 5 * 1024 * 1024
@@ -212,8 +224,11 @@ def close(self):
         self._http.close()
 
     def get_service_account_email(
-        self, project=None, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        project: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> str:
         """Get the email address of the project's BigQuery service account
 
         Note:
@@ -255,8 +270,12 @@ def get_service_account_email(
         return api_response["email"]
 
     def list_projects(
-        self, max_results=None, page_token=None, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        max_results: int = None,
+        page_token: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> page_iterator.Iterator:
         """List projects for the project associated with this client.
 
         See
@@ -309,14 +328,14 @@ def api_request(*args, **kwargs):
 
     def list_datasets(
         self,
-        project=None,
-        include_all=False,
-        filter=None,
-        max_results=None,
-        page_token=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        project: str = None,
+        include_all: bool = False,
+        filter: str = None,
+        max_results: int = None,
+        page_token: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> page_iterator.Iterator:
         """List datasets for the project associated with this client.
 
         See
@@ -386,7 +405,7 @@ def api_request(*args, **kwargs):
             extra_params=extra_params,
         )
 
-    def dataset(self, dataset_id, project=None):
+    def dataset(self, dataset_id: str, project: str = None) -> DatasetReference:
         """Deprecated: Construct a reference to a dataset.
 
         .. deprecated:: 1.24.0
@@ -446,8 +465,12 @@ def _create_bqstorage_client(self):
         return bigquery_storage.BigQueryReadClient(credentials=self._credentials)
 
     def create_dataset(
-        self, dataset, exists_ok=False, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        dataset: Union[str, Dataset, DatasetReference],
+        exists_ok: bool = False,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Dataset:
         """API call: create the dataset via a POST request.
 
         See
@@ -519,8 +542,12 @@ def create_dataset(
             return self.get_dataset(dataset.reference, retry=retry)
 
     def create_routine(
-        self, routine, exists_ok=False, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        routine: Routine,
+        exists_ok: bool = False,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Routine:
         """[Beta] Create a routine via a POST request.
 
         See
@@ -569,7 +596,13 @@ def create_routine(
                 raise
             return self.get_routine(routine.reference, retry=retry)
 
-    def create_table(self, table, exists_ok=False, retry=DEFAULT_RETRY, timeout=None):
+    def create_table(
+        self,
+        table: Union[str, Table, TableReference],
+        exists_ok: bool = False,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Table:
         """API call:  create a table via a PUT request
 
         See
@@ -636,7 +669,12 @@ def _call_api(
                 return call()
         return call()
 
-    def get_dataset(self, dataset_ref, retry=DEFAULT_RETRY, timeout=None):
+    def get_dataset(
+        self,
+        dataset_ref: Union[DatasetReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Dataset:
         """Fetch the dataset referenced by ``dataset_ref``
 
         Args:
@@ -675,8 +713,12 @@ def get_dataset(self, dataset_ref, retry=DEFAULT_RETRY, timeout=None):
         return Dataset.from_api_repr(api_response)
 
     def get_iam_policy(
-        self, table, requested_policy_version=1, retry=DEFAULT_RETRY, timeout=None,
-    ):
+        self,
+        table: Union[Table, TableReference],
+        requested_policy_version: int = 1,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Policy:
         if not isinstance(table, (Table, TableReference)):
             raise TypeError("table must be a Table or TableReference")
 
@@ -700,8 +742,13 @@ def get_iam_policy(
         return Policy.from_api_repr(response)
 
     def set_iam_policy(
-        self, table, policy, updateMask=None, retry=DEFAULT_RETRY, timeout=None,
-    ):
+        self,
+        table: Union[Table, TableReference],
+        policy: Policy,
+        updateMask: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Policy:
         if not isinstance(table, (Table, TableReference)):
             raise TypeError("table must be a Table or TableReference")
 
@@ -729,7 +776,11 @@ def set_iam_policy(
         return Policy.from_api_repr(response)
 
     def test_iam_permissions(
-        self, table, permissions, retry=DEFAULT_RETRY, timeout=None,
+        self,
+        table: Union[Table, TableReference],
+        permissions: Sequence[str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
     ):
         if not isinstance(table, (Table, TableReference)):
             raise TypeError("table must be a Table or TableReference")
@@ -750,7 +801,12 @@ def test_iam_permissions(
 
         return response
 
-    def get_model(self, model_ref, retry=DEFAULT_RETRY, timeout=None):
+    def get_model(
+        self,
+        model_ref: Union[ModelReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Model:
         """[Beta] Fetch the model referenced by ``model_ref``.
 
          Args:
@@ -788,7 +844,12 @@ def get_model(self, model_ref, retry=DEFAULT_RETRY, timeout=None):
         )
         return Model.from_api_repr(api_response)
 
-    def get_routine(self, routine_ref, retry=DEFAULT_RETRY, timeout=None):
+    def get_routine(
+        self,
+        routine_ref: Union[Routine, RoutineReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Routine:
         """[Beta] Get the routine referenced by ``routine_ref``.
 
          Args:
@@ -827,7 +888,12 @@ def get_routine(self, routine_ref, retry=DEFAULT_RETRY, timeout=None):
         )
         return Routine.from_api_repr(api_response)
 
-    def get_table(self, table, retry=DEFAULT_RETRY, timeout=None):
+    def get_table(
+        self,
+        table: Union[Table, TableReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Table:
         """Fetch the table referenced by ``table``.
 
         Args:
@@ -863,7 +929,13 @@ def get_table(self, table, retry=DEFAULT_RETRY, timeout=None):
         )
         return Table.from_api_repr(api_response)
 
-    def update_dataset(self, dataset, fields, retry=DEFAULT_RETRY, timeout=None):
+    def update_dataset(
+        self,
+        dataset: Dataset,
+        fields: Sequence[str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Dataset:
         """Change some fields of a dataset.
 
         Use ``fields`` to specify which fields to update. At least one field
@@ -927,7 +999,13 @@ def update_dataset(self, dataset, fields, retry=DEFAULT_RETRY, timeout=None):
         )
         return Dataset.from_api_repr(api_response)
 
-    def update_model(self, model, fields, retry=DEFAULT_RETRY, timeout=None):
+    def update_model(
+        self,
+        model: Model,
+        fields: Sequence[str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Model:
         """[Beta] Change some fields of a model.
 
         Use ``fields`` to specify which fields to update. At least one field
@@ -985,7 +1063,13 @@ def update_model(self, model, fields, retry=DEFAULT_RETRY, timeout=None):
         )
         return Model.from_api_repr(api_response)
 
-    def update_routine(self, routine, fields, retry=DEFAULT_RETRY, timeout=None):
+    def update_routine(
+        self,
+        routine: Routine,
+        fields: Sequence[str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Routine:
         """[Beta] Change some fields of a routine.
 
         Use ``fields`` to specify which fields to update. At least one field
@@ -1053,7 +1137,13 @@ def update_routine(self, routine, fields, retry=DEFAULT_RETRY, timeout=None):
         )
         return Routine.from_api_repr(api_response)
 
-    def update_table(self, table, fields, retry=DEFAULT_RETRY, timeout=None):
+    def update_table(
+        self,
+        table: Table,
+        fields: Sequence[str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Table:
         """Change some fields of a table.
 
         Use ``fields`` to specify which fields to update. At least one field
@@ -1114,12 +1204,12 @@ def update_table(self, table, fields, retry=DEFAULT_RETRY, timeout=None):
 
     def list_models(
         self,
-        dataset,
-        max_results=None,
-        page_token=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        dataset: Union[Dataset, DatasetReference, str],
+        max_results: int = None,
+        page_token: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> page_iterator.Iterator:
         """[Beta] List models in the dataset.
 
         See
@@ -1191,12 +1281,12 @@ def api_request(*args, **kwargs):
 
     def list_routines(
         self,
-        dataset,
-        max_results=None,
-        page_token=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        dataset: Union[Dataset, DatasetReference, str],
+        max_results: int = None,
+        page_token: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> page_iterator.Iterator:
         """[Beta] List routines in the dataset.
 
         See
@@ -1269,12 +1359,12 @@ def api_request(*args, **kwargs):
 
     def list_tables(
         self,
-        dataset,
-        max_results=None,
-        page_token=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        dataset: Union[Dataset, DatasetReference, str],
+        max_results: int = None,
+        page_token: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> page_iterator.Iterator:
         """List tables in the dataset.
 
         See
@@ -1346,12 +1436,12 @@ def api_request(*args, **kwargs):
 
     def delete_dataset(
         self,
-        dataset,
-        delete_contents=False,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-        not_found_ok=False,
-    ):
+        dataset: Union[Dataset, DatasetReference, str],
+        delete_contents: bool = False,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        not_found_ok: bool = False,
+    ) -> None:
         """Delete a dataset.
 
         See
@@ -1411,8 +1501,12 @@ def delete_dataset(
                 raise
 
     def delete_model(
-        self, model, retry=DEFAULT_RETRY, timeout=None, not_found_ok=False
-    ):
+        self,
+        model: Union[Model, ModelReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        not_found_ok: bool = False,
+    ) -> None:
         """[Beta] Delete a model
 
         See
@@ -1459,8 +1553,12 @@ def delete_model(
                 raise
 
     def delete_routine(
-        self, routine, retry=DEFAULT_RETRY, timeout=None, not_found_ok=False
-    ):
+        self,
+        routine: Union[Routine, RoutineReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        not_found_ok: bool = False,
+    ) -> None:
         """[Beta] Delete a routine.
 
         See
@@ -1509,8 +1607,12 @@ def delete_routine(
                 raise
 
     def delete_table(
-        self, table, retry=DEFAULT_RETRY, timeout=None, not_found_ok=False
-    ):
+        self,
+        table: Union[Table, TableReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        not_found_ok: bool = False,
+    ) -> None:
         """Delete a table
 
         See
@@ -1555,7 +1657,13 @@ def delete_table(
                 raise
 
     def _get_query_results(
-        self, job_id, retry, project=None, timeout_ms=None, location=None, timeout=None,
+        self,
+        job_id: str,
+        retry: retries.Retry,
+        project: str = None,
+        timeout_ms: int = None,
+        location: str = None,
+        timeout: float = None,
     ):
         """Get the query results object for a query job.
 
@@ -1609,7 +1717,7 @@ def _get_query_results(
         )
         return _QueryResults.from_api_repr(resource)
 
-    def job_from_resource(self, resource):
+    def job_from_resource(self, resource: dict) -> job.UnknownJob:
         """Detect correct job type from resource and instantiate.
 
         Args:
@@ -1635,7 +1743,12 @@ def job_from_resource(self, resource):
             return job.QueryJob.from_api_repr(resource, self)
         return job.UnknownJob.from_api_repr(resource, self)
 
-    def create_job(self, job_config, retry=DEFAULT_RETRY, timeout=None):
+    def create_job(
+        self,
+        job_config: dict,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Union[job.LoadJob, job.CopyJob, job.ExtractJob, job.QueryJob]:
         """Create a new job.
         Args:
             job_config (dict): configuration job representation returned from the API.
@@ -1726,8 +1839,13 @@ def create_job(self, job_config, retry=DEFAULT_RETRY, timeout=None):
             raise TypeError("Invalid job configuration received.")
 
     def get_job(
-        self, job_id, project=None, location=None, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        job_id: str,
+        project: str = None,
+        location: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Union[job.LoadJob, job.CopyJob, job.ExtractJob, job.QueryJob]:
         """Fetch a job for the project associated with this client.
 
         See
@@ -1783,8 +1901,13 @@ def get_job(
         return self.job_from_resource(resource)
 
     def cancel_job(
-        self, job_id, project=None, location=None, retry=DEFAULT_RETRY, timeout=None
-    ):
+        self,
+        job_id: str,
+        project: str = None,
+        location: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Union[job.LoadJob, job.CopyJob, job.ExtractJob, job.QueryJob]:
         """Attempt to cancel a job from a job ID.
 
         See
@@ -1841,17 +1964,17 @@ def cancel_job(
 
     def list_jobs(
         self,
-        project=None,
-        parent_job=None,
-        max_results=None,
-        page_token=None,
-        all_users=None,
-        state_filter=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-        min_creation_time=None,
-        max_creation_time=None,
-    ):
+        project: str = None,
+        parent_job: Union[_AsyncJob, str] = None,
+        max_results: int = None,
+        page_token: str = None,
+        all_users: bool = None,
+        state_filter: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        min_creation_time: datetime.datetime = None,
+        max_creation_time: datetime.datetime = None,
+    ) -> page_iterator.Iterator:
         """List jobs for the project associated with this client.
 
         See
@@ -1951,16 +2074,16 @@ def api_request(*args, **kwargs):
 
     def load_table_from_uri(
         self,
-        source_uris,
-        destination,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        source_uris: Union[str, Sequence[str]],
+        destination: Union[Table, TableReference, str],
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: LoadJobConfig = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> job.LoadJob:
         """Starts a job for loading data into a table from CloudStorage.
 
         See
@@ -2033,18 +2156,18 @@ def load_table_from_uri(
 
     def load_table_from_file(
         self,
-        file_obj,
-        destination,
-        rewind=False,
-        size=None,
-        num_retries=_DEFAULT_NUM_RETRIES,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        timeout=None,
-    ):
+        file_obj: BinaryIO,
+        destination: Union[Table, TableReference, str],
+        rewind: bool = False,
+        size: int = None,
+        num_retries: int = _DEFAULT_NUM_RETRIES,
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: LoadJobConfig = None,
+        timeout: float = None,
+    ) -> job.LoadJob:
         """Upload the contents of this table from a file-like object.
 
         Similar to :meth:`load_table_from_uri`, this method creates, starts and
@@ -2138,16 +2261,16 @@ def load_table_from_file(
     def load_table_from_dataframe(
         self,
         dataframe,
-        destination,
-        num_retries=_DEFAULT_NUM_RETRIES,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        parquet_compression="snappy",
-        timeout=None,
-    ):
+        destination: Union[Table, TableReference, str],
+        num_retries: int = _DEFAULT_NUM_RETRIES,
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: LoadJobConfig = None,
+        parquet_compression: str = "snappy",
+        timeout: float = None,
+    ) -> job.LoadJob:
         """Upload the contents of a table from a pandas DataFrame.
 
         Similar to :meth:`load_table_from_uri`, this method creates, starts and
@@ -2355,16 +2478,16 @@ def load_table_from_dataframe(
 
     def load_table_from_json(
         self,
-        json_rows,
-        destination,
-        num_retries=_DEFAULT_NUM_RETRIES,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        timeout=None,
-    ):
+        json_rows: Iterable[Dict[str, Any]],
+        destination: Union[Table, TableReference, str],
+        num_retries: int = _DEFAULT_NUM_RETRIES,
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: LoadJobConfig = None,
+        timeout: float = None,
+    ) -> job.LoadJob:
         """Upload the contents of a table from a JSON string or dict.
 
         Args:
@@ -2595,16 +2718,16 @@ def _do_multipart_upload(self, stream, metadata, size, num_retries, timeout):
 
     def copy_table(
         self,
-        sources,
-        destination,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        sources: Union[Table, TableReference, str],
+        destination: Union[Table, TableReference, str],
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: CopyJobConfig = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> job.CopyJob:
         """Copy one or more tables to another table.
 
         See
@@ -2698,17 +2821,17 @@ def copy_table(
 
     def extract_table(
         self,
-        source,
-        destination_uris,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        job_config=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-        source_type="Table",
-    ):
+        source: Union[Table, TableReference, Model, ModelReference, str],
+        destination_uris: Union[str, Sequence[str]],
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        job_config: ExtractJobConfig = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        source_type: str = "Table",
+    ) -> job.ExtractJob:
         """Start a job to extract a table into Cloud Storage files.
 
         See
@@ -2797,15 +2920,15 @@ def extract_table(
 
     def query(
         self,
-        query,
-        job_config=None,
-        job_id=None,
-        job_id_prefix=None,
-        location=None,
-        project=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        query: str,
+        job_config: QueryJobConfig = None,
+        job_id: str = None,
+        job_id_prefix: str = None,
+        location: str = None,
+        project: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> job.QueryJob:
         """Run a SQL query.
 
         See
@@ -2882,7 +3005,13 @@ def query(
 
         return query_job
 
-    def insert_rows(self, table, rows, selected_fields=None, **kwargs):
+    def insert_rows(
+        self,
+        table: Union[Table, TableReference, str],
+        rows: Union[Sequence[Tuple], Sequence[Dict]],
+        selected_fields: Sequence[SchemaField] = None,
+        **kwargs: dict,
+    ) -> Sequence[dict]:
         """Insert rows into a table via the streaming API.
 
         See
@@ -2945,8 +3074,13 @@ def insert_rows(self, table, rows, selected_fields=None, **kwargs):
         return self.insert_rows_json(table, json_rows, **kwargs)
 
     def insert_rows_from_dataframe(
-        self, table, dataframe, selected_fields=None, chunk_size=500, **kwargs
-    ):
+        self,
+        table: Union[Table, TableReference, str],
+        dataframe,
+        selected_fields: Sequence[SchemaField] = None,
+        chunk_size: int = 500,
+        **kwargs: Dict,
+    ) -> Sequence[Sequence[dict]]:
         """Insert rows into a table from a dataframe via the streaming API.
 
         Args:
@@ -2994,15 +3128,15 @@ def insert_rows_from_dataframe(
 
     def insert_rows_json(
         self,
-        table,
-        json_rows,
-        row_ids=None,
-        skip_invalid_rows=None,
-        ignore_unknown_values=None,
-        template_suffix=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        table: Union[Table, TableReference, str],
+        json_rows: Sequence[Dict],
+        row_ids: Sequence[str] = None,
+        skip_invalid_rows: bool = None,
+        ignore_unknown_values: bool = None,
+        template_suffix: str = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Sequence[dict]:
         """Insert rows into a table without applying local type conversions.
 
         See
@@ -3098,7 +3232,12 @@ def insert_rows_json(
 
         return errors
 
-    def list_partitions(self, table, retry=DEFAULT_RETRY, timeout=None):
+    def list_partitions(
+        self,
+        table: Union[Table, TableReference, str],
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> Sequence[str]:
         """List the partitions in a table.
 
         Args:
@@ -3140,15 +3279,15 @@ def list_partitions(self, table, retry=DEFAULT_RETRY, timeout=None):
 
     def list_rows(
         self,
-        table,
-        selected_fields=None,
-        max_results=None,
-        page_token=None,
-        start_index=None,
-        page_size=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        table: Union[Table, TableListItem, TableReference, str],
+        selected_fields: Sequence[SchemaField] = None,
+        max_results: int = None,
+        page_token: str = None,
+        start_index: int = None,
+        page_size: int = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> RowIterator:
         """List the rows of the table.
 
         See
@@ -3249,18 +3388,18 @@ def list_rows(
 
     def _list_rows_from_query_results(
         self,
-        job_id,
-        location,
-        project,
-        schema,
-        total_rows=None,
-        destination=None,
-        max_results=None,
-        start_index=None,
-        page_size=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-    ):
+        job_id: str,
+        location: str,
+        project: str,
+        schema: SchemaField,
+        total_rows: int = None,
+        destination: Union[Table, TableReference, TableListItem, str] = None,
+        max_results: int = None,
+        start_index: int = None,
+        page_size: int = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+    ) -> RowIterator:
         """List the rows of a completed query.
         See
         https://cloud.google.com/bigquery/docs/reference/rest/v2/jobs/getQueryResults
@@ -3340,7 +3479,7 @@ def _schema_to_json_file_object(self, schema_list, file_obj):
         """
         json.dump(schema_list, file_obj, indent=2, sort_keys=True)
 
-    def schema_from_json(self, file_or_path):
+    def schema_from_json(self, file_or_path: Union[str, BinaryIO]):
         """Takes a file object or file path that contains json that describes
         a table schema.
 
@@ -3353,7 +3492,9 @@ def schema_from_json(self, file_or_path):
         with open(file_or_path) as file_obj:
             return self._schema_from_json_file_object(file_obj)
 
-    def schema_to_json(self, schema_list, destination):
+    def schema_to_json(
+        self, schema_list: Sequence[SchemaField], destination: Union[str, BinaryIO]
+    ):
         """Takes a list of schema field objects.
 
         Serializes the list of schema field objects as json to a file.
diff --git a/google/cloud/bigquery/dataset.py b/google/cloud/bigquery/dataset.py
index ce07c8048..81a0b8336 100644
--- a/google/cloud/bigquery/dataset.py
+++ b/google/cloud/bigquery/dataset.py
@@ -19,7 +19,7 @@
 import six
 import copy
 
-import google.cloud._helpers
+import google.cloud._helpers  # type: ignore
 
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery.model import ModelReference
@@ -221,7 +221,7 @@ def to_api_repr(self):
         return resource
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "AccessEntry":
         """Factory: construct an access entry given its API representation
 
         Args:
@@ -289,7 +289,7 @@ def path(self):
     routine = _get_routine_reference
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "DatasetReference":
         """Factory: construct a dataset reference given its API representation
 
         Args:
@@ -305,7 +305,9 @@ def from_api_repr(cls, resource):
         return cls(project, dataset_id)
 
     @classmethod
-    def from_string(cls, dataset_id, default_project=None):
+    def from_string(
+        cls, dataset_id: str, default_project: str = None
+    ) -> "DatasetReference":
         """Construct a dataset reference from dataset ID string.
 
         Args:
@@ -351,7 +353,7 @@ def from_string(cls, dataset_id, default_project=None):
 
         return cls(output_project_id, output_dataset_id)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this dataset reference
 
         Returns:
@@ -641,7 +643,7 @@ def default_encryption_configuration(self, value):
         self._properties["defaultEncryptionConfiguration"] = api_repr
 
     @classmethod
-    def from_string(cls, full_dataset_id):
+    def from_string(cls, full_dataset_id: str) -> "Dataset":
         """Construct a dataset from fully-qualified dataset ID.
 
         Args:
@@ -665,7 +667,7 @@ def from_string(cls, full_dataset_id):
         return cls(DatasetReference.from_string(full_dataset_id))
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "Dataset":
         """Factory: construct a dataset given its API representation
 
         Args:
@@ -690,7 +692,7 @@ def from_api_repr(cls, resource):
         dataset._properties = copy.deepcopy(resource)
         return dataset
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this dataset
 
         Returns:
diff --git a/google/cloud/bigquery/dbapi/cursor.py b/google/cloud/bigquery/dbapi/cursor.py
index f48b47c12..2e4be5890 100644
--- a/google/cloud/bigquery/dbapi/cursor.py
+++ b/google/cloud/bigquery/dbapi/cursor.py
@@ -24,7 +24,7 @@
 from google.cloud.bigquery import job
 from google.cloud.bigquery.dbapi import _helpers
 from google.cloud.bigquery.dbapi import exceptions
-import google.cloud.exceptions
+import google.cloud.exceptions  # type: ignore
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -249,7 +249,7 @@ def _bqstorage_fetch(self, bqstorage_client):
         """
         # Hitting this code path with a BQ Storage client instance implies that
         # bigquery_storage can indeed be imported here without errors.
-        from google.cloud import bigquery_storage
+        from google.cloud import bigquery_storage  # type: ignore
 
         table_reference = self._query_job.destination
 
diff --git a/google/cloud/bigquery/external_config.py b/google/cloud/bigquery/external_config.py
index 112dfdba4..8a519d8a2 100644
--- a/google/cloud/bigquery/external_config.py
+++ b/google/cloud/bigquery/external_config.py
@@ -149,7 +149,7 @@ def type_(self):
     def type_(self, value):
         self._properties["type"] = value
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -159,7 +159,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "BigtableColumn":
         """Factory: construct a :class:`~.external_config.BigtableColumn`
         instance given its API representation.
 
@@ -251,7 +251,7 @@ def columns(self):
     def columns(self, value):
         self._properties["columns"] = [col.to_api_repr() for col in value]
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -261,7 +261,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "BigtableColumnFamily":
         """Factory: construct a :class:`~.external_config.BigtableColumnFamily`
         instance given its API representation.
 
@@ -333,7 +333,7 @@ def column_families(self):
     def column_families(self, value):
         self._properties["columnFamilies"] = [cf.to_api_repr() for cf in value]
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -343,7 +343,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "BigtableOptions":
         """Factory: construct a :class:`~.external_config.BigtableOptions`
         instance given its API representation.
 
@@ -450,7 +450,7 @@ def skip_leading_rows(self):
     def skip_leading_rows(self, value):
         self._properties["skipLeadingRows"] = str(value)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -459,7 +459,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "CSVOptions":
         """Factory: construct a :class:`~.external_config.CSVOptions` instance
         given its API representation.
 
@@ -513,7 +513,7 @@ def range(self):
     def range(self, value):
         self._properties["range"] = value
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -522,7 +522,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "GoogleSheetsOptions":
         """Factory: construct a :class:`~.external_config.GoogleSheetsOptions`
         instance given its API representation.
 
@@ -601,7 +601,7 @@ def require_partition_filter(self):
     def require_partition_filter(self, value):
         self._properties["requirePartitionFilter"] = value
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -610,7 +610,7 @@ def to_api_repr(self):
         return copy.deepcopy(self._properties)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "HivePartitioningOptions":
         """Factory: construct a :class:`~.external_config.HivePartitioningOptions`
         instance given its API representation.
 
@@ -767,7 +767,7 @@ def schema(self, value):
             prop = {"fields": [field.to_api_repr() for field in value]}
         self._properties["schema"] = prop
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of this object.
 
         Returns:
@@ -782,7 +782,7 @@ def to_api_repr(self):
         return config
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "ExternalConfig":
         """Factory: construct an :class:`~.external_config.ExternalConfig`
         instance given its API representation.
 
diff --git a/google/cloud/bigquery/job/base.py b/google/cloud/bigquery/job/base.py
index 2f4ae1460..f11537cbe 100644
--- a/google/cloud/bigquery/job/base.py
+++ b/google/cloud/bigquery/job/base.py
@@ -17,13 +17,16 @@
 import copy
 import threading
 
-from google.api_core import exceptions
-import google.api_core.future.polling
+from google.api_core import exceptions  # type: ignore
+import google.api_core.future.polling  # type: ignore
 from six.moves import http_client
 
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery.retry import DEFAULT_RETRY
 
+# Types needed only for Type Hints
+from google.api_core import retry as retries  # type: ignore
+
 
 _DONE_STATE = "DONE"
 _STOPPED_REASON = "stopped"
@@ -439,7 +442,9 @@ def _begin(self, client=None, retry=DEFAULT_RETRY, timeout=None):
         )
         self._set_properties(api_response)
 
-    def exists(self, client=None, retry=DEFAULT_RETRY, timeout=None):
+    def exists(
+        self, client=None, retry: retries.Retry = DEFAULT_RETRY, timeout: float = None
+    ) -> bool:
         """API call:  test for the existence of the job via a GET request
 
         See
@@ -482,7 +487,9 @@ def exists(self, client=None, retry=DEFAULT_RETRY, timeout=None):
         else:
             return True
 
-    def reload(self, client=None, retry=DEFAULT_RETRY, timeout=None):
+    def reload(
+        self, client=None, retry: retries.Retry = DEFAULT_RETRY, timeout: float = None
+    ):
         """API call:  refresh job properties via a GET request.
 
         See
@@ -517,7 +524,9 @@ def reload(self, client=None, retry=DEFAULT_RETRY, timeout=None):
         )
         self._set_properties(api_response)
 
-    def cancel(self, client=None, retry=DEFAULT_RETRY, timeout=None):
+    def cancel(
+        self, client=None, retry: retries.Retry = DEFAULT_RETRY, timeout: float = None
+    ) -> bool:
         """API call:  cancel job via a POST request
 
         See
@@ -583,7 +592,12 @@ def _set_future_result(self):
             else:
                 self.set_result(self)
 
-    def done(self, retry=DEFAULT_RETRY, timeout=None, reload=True):
+    def done(
+        self,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        reload: bool = True,
+    ) -> bool:
         """Checks if the job is complete.
 
         Args:
@@ -604,7 +618,9 @@ def done(self, retry=DEFAULT_RETRY, timeout=None, reload=True):
             self.reload(retry=retry, timeout=timeout)
         return self.state == _DONE_STATE
 
-    def result(self, retry=DEFAULT_RETRY, timeout=None):
+    def result(
+        self, retry: retries.Retry = DEFAULT_RETRY, timeout: float = None
+    ) -> "_AsyncJob":
         """Start the job and wait for it to complete and get the result.
 
         Args:
@@ -750,7 +766,7 @@ def _del_sub_prop(self, key):
         """
         _helpers._del_sub_prop(self._properties, [self._job_type, key])
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of the job config.
 
         Returns:
@@ -780,7 +796,7 @@ def _fill_from_default(self, default_job_config):
                 + repr(default_job_config._job_type)
             )
 
-        new_job_config = self.__class__()
+        new_job_config = self.__class__()  # type: ignore
 
         default_job_properties = copy.deepcopy(default_job_config._properties)
         for key in self._properties:
@@ -793,7 +809,7 @@ def _fill_from_default(self, default_job_config):
         return new_job_config
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "_JobConfig":
         """Factory: construct a job configuration given its API representation
 
         Args:
@@ -804,7 +820,7 @@ def from_api_repr(cls, resource):
         Returns:
             google.cloud.bigquery.job._JobConfig: Configuration parsed from ``resource``.
         """
-        job_config = cls()
+        job_config = cls()  # type: ignore
         job_config._properties = resource
         return job_config
 
@@ -891,7 +907,7 @@ class UnknownJob(_AsyncJob):
     """A job whose type cannot be determined."""
 
     @classmethod
-    def from_api_repr(cls, resource, client):
+    def from_api_repr(cls, resource: dict, client) -> "UnknownJob":
         """Construct an UnknownJob from the JSON representation.
 
         Args:
diff --git a/google/cloud/bigquery/job/extract.py b/google/cloud/bigquery/job/extract.py
index a6e262a32..3373bcdef 100644
--- a/google/cloud/bigquery/job/extract.py
+++ b/google/cloud/bigquery/job/extract.py
@@ -241,7 +241,7 @@ def to_api_repr(self):
         }
 
     @classmethod
-    def from_api_repr(cls, resource, client):
+    def from_api_repr(cls, resource: dict, client) -> "ExtractJob":
         """Factory:  construct a job given its API representation
 
         .. note:
diff --git a/google/cloud/bigquery/job/load.py b/google/cloud/bigquery/job/load.py
index e784af0a6..b8174af3e 100644
--- a/google/cloud/bigquery/job/load.py
+++ b/google/cloud/bigquery/job/load.py
@@ -733,7 +733,7 @@ def to_api_repr(self):
         }
 
     @classmethod
-    def from_api_repr(cls, resource, client):
+    def from_api_repr(cls, resource: dict, client) -> "LoadJob":
         """Factory:  construct a job given its API representation
 
         .. note:
diff --git a/google/cloud/bigquery/job/query.py b/google/cloud/bigquery/job/query.py
index 9e8908613..1ac5c063b 100644
--- a/google/cloud/bigquery/job/query.py
+++ b/google/cloud/bigquery/job/query.py
@@ -47,6 +47,12 @@
 from google.cloud.bigquery.job.base import _JobConfig
 from google.cloud.bigquery.job.base import _JobReference
 
+# Types needed only for Type Hints
+from google.api_core import retry as retries  # type: ignore
+from google.cloud import bigquery_storage  # type: ignore
+from google.cloud.bigquery.table import RowIterator
+from typing import Any, Dict, Union
+
 
 _CONTAINS_ORDER_BY = re.compile(r"ORDER\s+BY", re.IGNORECASE)
 _TIMEOUT_BUFFER_SECS = 0.1
@@ -492,7 +498,7 @@ def schema_update_options(self):
     def schema_update_options(self, values):
         self._set_sub_prop("schemaUpdateOptions", values)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Build an API representation of the query job config.
 
         Returns:
@@ -719,7 +725,7 @@ def to_api_repr(self):
         }
 
     @classmethod
-    def from_api_repr(cls, resource, client):
+    def from_api_repr(cls, resource: dict, client) -> "QueryJob":
         """Factory:  construct a job given its API representation
 
         Args:
@@ -975,7 +981,12 @@ def estimated_bytes_processed(self):
             result = int(result)
         return result
 
-    def done(self, retry=DEFAULT_RETRY, timeout=None, reload=True):
+    def done(
+        self,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        reload: bool = True,
+    ) -> bool:
         """Refresh the job and checks if it is complete.
 
         Args:
@@ -1073,7 +1084,9 @@ def _begin(self, client=None, retry=DEFAULT_RETRY, timeout=None):
             exc.query_job = self
             raise
 
-    def _reload_query_results(self, retry=DEFAULT_RETRY, timeout=None):
+    def _reload_query_results(
+        self, retry: retries.Retry = DEFAULT_RETRY, timeout: float = None
+    ):
         """Refresh the cached query results.
 
         Args:
@@ -1114,12 +1127,12 @@ def _reload_query_results(self, retry=DEFAULT_RETRY, timeout=None):
 
     def result(
         self,
-        page_size=None,
-        max_results=None,
-        retry=DEFAULT_RETRY,
-        timeout=None,
-        start_index=None,
-    ):
+        page_size: int = None,
+        max_results: int = None,
+        retry: retries.Retry = DEFAULT_RETRY,
+        timeout: float = None,
+        start_index: int = None,
+    ) -> Union[RowIterator, _EmptyRowIterator]:
         """Start the job and wait for it to complete and get the result.
 
         Args:
@@ -1197,10 +1210,10 @@ def result(
     # changes to table.RowIterator.to_arrow()
     def to_arrow(
         self,
-        progress_bar_type=None,
-        bqstorage_client=None,
-        create_bqstorage_client=True,
-    ):
+        progress_bar_type: str = None,
+        bqstorage_client: bigquery_storage.BigQueryReadClient = None,
+        create_bqstorage_client: bool = True,
+    ) -> Any:
         """[Beta] Create a class:`pyarrow.Table` by loading all pages of a
         table or query.
 
@@ -1266,12 +1279,12 @@ def to_arrow(
     # changes to table.RowIterator.to_dataframe()
     def to_dataframe(
         self,
-        bqstorage_client=None,
-        dtypes=None,
-        progress_bar_type=None,
-        create_bqstorage_client=True,
-        date_as_object=True,
-    ):
+        bqstorage_client: bigquery_storage.BigQueryReadClient = None,
+        dtypes: Dict[str, Any] = None,
+        progress_bar_type: str = None,
+        create_bqstorage_client: bool = True,
+        date_as_object: bool = True,
+    ) -> Any:
         """Return a pandas DataFrame from a QueryJob
 
         Args:
@@ -1351,7 +1364,7 @@ def __init__(self, kind, substeps):
         self.substeps = list(substeps)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "QueryPlanEntryStep":
         """Factory: construct instance from the JSON repr.
 
         Args:
@@ -1381,7 +1394,7 @@ def __init__(self):
         self._properties = {}
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "QueryPlanEntry":
         """Factory: construct instance from the JSON repr.
 
         Args:
diff --git a/google/cloud/bigquery/magics/line_arg_parser/lexer.py b/google/cloud/bigquery/magics/line_arg_parser/lexer.py
index 17e1ffdae..55a882b27 100644
--- a/google/cloud/bigquery/magics/line_arg_parser/lexer.py
+++ b/google/cloud/bigquery/magics/line_arg_parser/lexer.py
@@ -149,7 +149,7 @@ def __prepare__(metacls, name, bases, **kwargs):
     def __new__(metacls, name, bases, oldclassdict):
         # Scan through the declared enum members and convert any value that is a plain
         # empty tuple into a `str` of the name instead.
-        newclassdict = enum._EnumDict()
+        newclassdict = enum._EnumDict()  # type: ignore
         for key, val in oldclassdict.items():
             if val == ():
                 val = key
diff --git a/google/cloud/bigquery/model.py b/google/cloud/bigquery/model.py
index 0f5d8f83b..b441f491e 100644
--- a/google/cloud/bigquery/model.py
+++ b/google/cloud/bigquery/model.py
@@ -21,8 +21,8 @@
 from google.protobuf import json_format
 import six
 
-import google.cloud._helpers
-from google.api_core import datetime_helpers
+import google.cloud._helpers  # type: ignore
+from google.api_core import datetime_helpers  # type: ignore
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery_v2 import types
 from google.cloud.bigquery.encryption_configuration import EncryptionConfiguration
@@ -280,7 +280,7 @@ def encryption_configuration(self, value):
         self._properties["encryptionConfiguration"] = api_repr
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "Model":
         """Factory: construct a model resource given its API representation
 
         Args:
@@ -323,7 +323,7 @@ def _build_resource(self, filter_fields):
     def __repr__(self):
         return "Model(reference={})".format(repr(self.reference))
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this model.
 
         Returns:
@@ -390,7 +390,9 @@ def from_api_repr(cls, resource):
         return ref
 
     @classmethod
-    def from_string(cls, model_id, default_project=None):
+    def from_string(
+        cls, model_id: str, default_project: str = None
+    ) -> "ModelReference":
         """Construct a model reference from model ID string.
 
         Args:
@@ -418,7 +420,7 @@ def from_string(cls, model_id, default_project=None):
             {"projectId": proj, "datasetId": dset, "modelId": model}
         )
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this model reference.
 
         Returns:
diff --git a/google/cloud/bigquery/opentelemetry_tracing.py b/google/cloud/bigquery/opentelemetry_tracing.py
index b9d18efad..ea2ac92cf 100644
--- a/google/cloud/bigquery/opentelemetry_tracing.py
+++ b/google/cloud/bigquery/opentelemetry_tracing.py
@@ -14,13 +14,13 @@
 
 import logging
 from contextlib import contextmanager
-from google.api_core.exceptions import GoogleAPICallError
+from google.api_core.exceptions import GoogleAPICallError  # type: ignore
 
 logger = logging.getLogger(__name__)
 try:
-    from opentelemetry import trace
-    from opentelemetry.instrumentation.utils import http_status_to_canonical_code
-    from opentelemetry.trace.status import Status
+    from opentelemetry import trace  # type: ignore
+    from opentelemetry.instrumentation.utils import http_status_to_canonical_code  # type: ignore
+    from opentelemetry.trace.status import Status  # type: ignore
 
     HAS_OPENTELEMETRY = True
 
diff --git a/google/cloud/bigquery/query.py b/google/cloud/bigquery/query.py
index f2ed6337e..239ecb484 100644
--- a/google/cloud/bigquery/query.py
+++ b/google/cloud/bigquery/query.py
@@ -53,7 +53,7 @@ class _AbstractQueryParameter(object):
     """
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "ScalarQueryParameter":
         """Factory: construct parameter from JSON resource.
 
         Args:
@@ -64,7 +64,7 @@ def from_api_repr(cls, resource):
         """
         raise NotImplementedError
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct JSON API representation for the parameter.
 
         Returns:
@@ -96,7 +96,7 @@ def __init__(self, name, type_, value):
         self.value = value
 
     @classmethod
-    def positional(cls, type_, value):
+    def positional(cls, type_: str, value) -> "ScalarQueryParameter":
         """Factory for positional paramater.
 
         Args:
@@ -114,7 +114,7 @@ def positional(cls, type_, value):
         return cls(None, type_, value)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "ScalarQueryParameter":
         """Factory: construct parameter from JSON resource.
 
         Args:
@@ -136,7 +136,7 @@ def from_api_repr(cls, resource):
 
         return cls(name, type_, converted)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct JSON API representation for the parameter.
 
         Returns:
@@ -197,7 +197,7 @@ def __init__(self, name, array_type, values):
         self.values = values
 
     @classmethod
-    def positional(cls, array_type, values):
+    def positional(cls, array_type: str, values: list) -> "ArrayQueryParameter":
         """Factory for positional parameters.
 
         Args:
@@ -242,7 +242,7 @@ def _from_api_repr_scalar(cls, resource):
         return cls(name, array_type, converted)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "ArrayQueryParameter":
         """Factory: construct parameter from JSON resource.
 
         Args:
@@ -256,7 +256,7 @@ def from_api_repr(cls, resource):
             return cls._from_api_repr_struct(resource)
         return cls._from_api_repr_scalar(resource)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct JSON API representation for the parameter.
 
         Returns:
@@ -350,7 +350,7 @@ def positional(cls, *sub_params):
         return cls(None, *sub_params)
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "StructQueryParameter":
         """Factory: construct parameter from JSON resource.
 
         Args:
@@ -390,7 +390,7 @@ def from_api_repr(cls, resource):
             instance.struct_values[key] = converted
         return instance
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct JSON API representation for the parameter.
 
         Returns:
diff --git a/google/cloud/bigquery/retry.py b/google/cloud/bigquery/retry.py
index 4bc4b757f..7635fcc0a 100644
--- a/google/cloud/bigquery/retry.py
+++ b/google/cloud/bigquery/retry.py
@@ -12,8 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-from google.api_core import exceptions
-from google.api_core import retry
+from google.api_core import exceptions  # type: ignore
+from google.api_core import retry  # type: ignore
 
 
 _RETRYABLE_REASONS = frozenset(
diff --git a/google/cloud/bigquery/routine.py b/google/cloud/bigquery/routine.py
index 6a0ed9fb0..c2509b1e4 100644
--- a/google/cloud/bigquery/routine.py
+++ b/google/cloud/bigquery/routine.py
@@ -119,12 +119,14 @@ def created(self):
 
         Read-only.
         """
+        # pytype: disable=module-attr
         value = self._properties.get(self._PROPERTY_TO_API_FIELD["created"])
         if value is not None and value != 0:
             # value will be in milliseconds.
             return google.cloud._helpers._datetime_from_microseconds(
                 1000.0 * float(value)
             )
+        # pytype: enable=module-attr
 
     @property
     def modified(self):
@@ -133,12 +135,14 @@ def modified(self):
 
         Read-only.
         """
+        # pytype: disable=module-attr
         value = self._properties.get(self._PROPERTY_TO_API_FIELD["modified"])
         if value is not None and value != 0:
             # value will be in milliseconds.
             return google.cloud._helpers._datetime_from_microseconds(
                 1000.0 * float(value)
             )
+        # pytype: enable=module-attr
 
     @property
     def language(self):
@@ -255,7 +259,7 @@ def description(self, value):
         self._properties[self._PROPERTY_TO_API_FIELD["description"]] = value
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "Routine":
         """Factory: construct a routine given its API representation.
 
         Args:
@@ -270,7 +274,7 @@ def from_api_repr(cls, resource):
         ref._properties = resource
         return ref
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this routine.
 
         Returns:
@@ -376,7 +380,7 @@ def data_type(self, value):
         self._properties[self._PROPERTY_TO_API_FIELD["data_type"]] = resource
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "RoutineArgument":
         """Factory: construct a routine argument given its API representation.
 
         Args:
@@ -390,7 +394,7 @@ def from_api_repr(cls, resource):
         ref._properties = resource
         return ref
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this routine argument.
 
         Returns:
@@ -427,17 +431,17 @@ def __init__(self):
     @property
     def project(self):
         """str: ID of the project containing the routine."""
-        return self._properties["projectId"]
+        return self._properties["projectId"]  # pytype: disable=key-error
 
     @property
     def dataset_id(self):
         """str: ID of dataset containing the routine."""
-        return self._properties["datasetId"]
+        return self._properties["datasetId"]  # pytype: disable=key-error
 
     @property
     def routine_id(self):
         """str: The routine ID."""
-        return self._properties["routineId"]
+        return self._properties["routineId"]  # pytype: disable=key-error
 
     @property
     def path(self):
@@ -449,7 +453,7 @@ def path(self):
         )
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "RoutineReference":
         """Factory: construct a routine reference given its API representation.
 
         Args:
@@ -465,7 +469,9 @@ def from_api_repr(cls, resource):
         return ref
 
     @classmethod
-    def from_string(cls, routine_id, default_project=None):
+    def from_string(
+        cls, routine_id: str, default_project: str = None
+    ) -> "RoutineReference":
         """Factory: construct a routine reference from routine ID string.
 
         Args:
@@ -493,7 +499,7 @@ def from_string(cls, routine_id, default_project=None):
             {"projectId": proj, "datasetId": dset, "routineId": routine}
         )
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this routine reference.
 
         Returns:
diff --git a/google/cloud/bigquery/schema.py b/google/cloud/bigquery/schema.py
index 8ae0a3a85..b65f630d6 100644
--- a/google/cloud/bigquery/schema.py
+++ b/google/cloud/bigquery/schema.py
@@ -84,7 +84,7 @@ def __init__(
         self._policy_tags = policy_tags
 
     @classmethod
-    def from_api_repr(cls, api_repr):
+    def from_api_repr(cls, api_repr: dict) -> "SchemaField":
         """Return a ``SchemaField`` object deserialized from a dictionary.
 
         Args:
@@ -157,7 +157,7 @@ def policy_tags(self):
         """
         return self._policy_tags
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Return a dictionary representing this schema field.
 
         Returns:
@@ -200,7 +200,7 @@ def _key(self):
             self._policy_tags,
         )
 
-    def to_standard_sql(self):
+    def to_standard_sql(self) -> types.StandardSqlField:
         """Return the field as the standard SQL field representation object.
 
         Returns:
@@ -375,7 +375,7 @@ def __repr__(self):
         return "PolicyTagList{}".format(self._key())
 
     @classmethod
-    def from_api_repr(cls, api_repr):
+    def from_api_repr(cls, api_repr: dict) -> "PolicyTagList":
         """Return a :class:`PolicyTagList` object deserialized from a dict.
 
         This method creates a new ``PolicyTagList`` instance that points to
@@ -398,7 +398,7 @@ def from_api_repr(cls, api_repr):
         names = api_repr.get("names", ())
         return cls(names=names)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Return a dictionary representing this object.
 
         This method returns the properties dict of the ``PolicyTagList``
diff --git a/google/cloud/bigquery/table.py b/google/cloud/bigquery/table.py
index 6daccf518..ccd8321ad 100644
--- a/google/cloud/bigquery/table.py
+++ b/google/cloud/bigquery/table.py
@@ -27,19 +27,19 @@
 import six
 
 try:
-    import pandas
+    import pandas  # type: ignore
 except ImportError:  # pragma: NO COVER
     pandas = None
 
 try:
-    import pyarrow
+    import pyarrow  # type: ignore
 except ImportError:  # pragma: NO COVER
     pyarrow = None
 
 import google.api_core.exceptions
-from google.api_core.page_iterator import HTTPIterator
+from google.api_core.page_iterator import HTTPIterator  # type: ignore
 
-import google.cloud._helpers
+import google.cloud._helpers  # type: ignore
 from google.cloud.bigquery import _helpers
 from google.cloud.bigquery import _pandas_helpers
 from google.cloud.bigquery.schema import _build_schema_resource
@@ -49,6 +49,10 @@
 from google.cloud.bigquery.external_config import ExternalConfig
 from google.cloud.bigquery.encryption_configuration import EncryptionConfiguration
 
+# Types needed only for Type Hints
+from google.cloud import bigquery_storage  # type: ignore
+from typing import Any, Dict, Iterable, Tuple
+
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -145,7 +149,9 @@ def path(self):
         )
 
     @classmethod
-    def from_string(cls, table_id, default_project=None):
+    def from_string(
+        cls, table_id: str, default_project: str = None
+    ) -> "TableReference":
         """Construct a table reference from table ID string.
 
         Args:
@@ -184,7 +190,7 @@ def from_string(cls, table_id, default_project=None):
         )
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "TableReference":
         """Factory:  construct a table reference given its API representation
 
         Args:
@@ -202,7 +208,7 @@ def from_api_repr(cls, resource):
         table_id = resource["tableId"]
         return cls(DatasetReference(project, dataset_id), table_id)
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Construct the API resource representation of this table reference.
 
         Returns:
@@ -214,7 +220,7 @@ def to_api_repr(self):
             "tableId": self._table_id,
         }
 
-    def to_bqstorage(self):
+    def to_bqstorage(self) -> str:
         """Construct a BigQuery Storage API representation of this table.
 
         Install the ``google-cloud-bigquery-storage`` package to use this
@@ -849,7 +855,7 @@ def external_data_configuration(self, value):
         self._properties["externalDataConfiguration"] = api_repr
 
     @classmethod
-    def from_string(cls, full_table_id):
+    def from_string(cls, full_table_id: str) -> "Table":
         """Construct a table from fully-qualified table ID.
 
         Args:
@@ -873,7 +879,7 @@ def from_string(cls, full_table_id):
         return cls(TableReference.from_string(full_table_id))
 
     @classmethod
-    def from_api_repr(cls, resource):
+    def from_api_repr(cls, resource: dict) -> "Table":
         """Factory: construct a table given its API representation
 
         Args:
@@ -909,7 +915,7 @@ def from_api_repr(cls, resource):
 
         return table
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Constructs the API resource of this table
 
         Returns:
@@ -917,7 +923,7 @@ def to_api_repr(self):
         """
         return copy.deepcopy(self._properties)
 
-    def to_bqstorage(self):
+    def to_bqstorage(self) -> str:
         """Construct a BigQuery Storage API representation of this table.
 
         Returns:
@@ -1106,7 +1112,7 @@ def clustering_fields(self):
             return list(prop.get("fields", ()))
 
     @classmethod
-    def from_string(cls, full_table_id):
+    def from_string(cls, full_table_id: str) -> "TableListItem":
         """Construct a table from fully-qualified table ID.
 
         Args:
@@ -1131,7 +1137,7 @@ def from_string(cls, full_table_id):
             {"tableReference": TableReference.from_string(full_table_id).to_api_repr()}
         )
 
-    def to_bqstorage(self):
+    def to_bqstorage(self) -> str:
         """Construct a BigQuery Storage API representation of this table.
 
         Returns:
@@ -1139,7 +1145,7 @@ def to_bqstorage(self):
         """
         return self.reference.to_bqstorage()
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Constructs the API resource of this table
 
         Returns:
@@ -1233,7 +1239,7 @@ def values(self):
         """
         return copy.deepcopy(self._xxx_values)
 
-    def keys(self):
+    def keys(self) -> Iterable[str]:
         """Return the keys for using a row as a dict.
 
         Returns:
@@ -1246,7 +1252,7 @@ def keys(self):
         """
         return six.iterkeys(self._xxx_field_to_index)
 
-    def items(self):
+    def items(self) -> Iterable[Tuple[str, Any]]:
         """Return items as ``(key, value)`` pairs.
 
         Returns:
@@ -1261,7 +1267,7 @@ def items(self):
         for key, index in six.iteritems(self._xxx_field_to_index):
             yield (key, copy.deepcopy(self._xxx_values[index]))
 
-    def get(self, key, default=None):
+    def get(self, key: str, default: Any = None) -> Any:
         """Return a value for key, with a default value if it does not exist.
 
         Args:
@@ -1523,10 +1529,10 @@ def _to_arrow_iterable(self, bqstorage_client=None):
     # changes to job.QueryJob.to_arrow()
     def to_arrow(
         self,
-        progress_bar_type=None,
-        bqstorage_client=None,
-        create_bqstorage_client=True,
-    ):
+        progress_bar_type: str = None,
+        bqstorage_client: bigquery_storage.BigQueryReadClient = None,
+        create_bqstorage_client: bool = True,
+    ) -> Any:
         """[Beta] Create a class:`pyarrow.Table` by loading all pages of a
         table or query.
 
@@ -1624,7 +1630,11 @@ def to_arrow(
             arrow_schema = _pandas_helpers.bq_to_arrow_schema(self._schema)
             return pyarrow.Table.from_batches(record_batches, schema=arrow_schema)
 
-    def to_dataframe_iterable(self, bqstorage_client=None, dtypes=None):
+    def to_dataframe_iterable(
+        self,
+        bqstorage_client: bigquery_storage.BigQueryReadClient = None,
+        dtypes: Dict[str, Any] = None,
+    ) -> Any:
         """Create an iterable of pandas DataFrames, to process the table as a stream.
 
         Args:
@@ -1684,12 +1694,12 @@ def to_dataframe_iterable(self, bqstorage_client=None, dtypes=None):
     # changes to job.QueryJob.to_dataframe()
     def to_dataframe(
         self,
-        bqstorage_client=None,
-        dtypes=None,
-        progress_bar_type=None,
-        create_bqstorage_client=True,
-        date_as_object=True,
-    ):
+        bqstorage_client: bigquery_storage.BigQueryReadClient = None,
+        dtypes: Dict[str, Any] = None,
+        progress_bar_type: str = None,
+        create_bqstorage_client: bool = True,
+        date_as_object: bool = True,
+    ) -> Any:
         """Create a pandas DataFrame by loading all pages of a query.
 
         Args:
@@ -1817,7 +1827,7 @@ def to_arrow(
         progress_bar_type=None,
         bqstorage_client=None,
         create_bqstorage_client=True,
-    ):
+    ) -> Any:
         """[Beta] Create an empty class:`pyarrow.Table`.
 
         Args:
@@ -1839,7 +1849,7 @@ def to_dataframe(
         progress_bar_type=None,
         create_bqstorage_client=True,
         date_as_object=True,
-    ):
+    ) -> Any:
         """Create an empty dataframe.
 
         Args:
@@ -2150,7 +2160,7 @@ def require_partition_filter(self, value):
         self._properties["requirePartitionFilter"] = value
 
     @classmethod
-    def from_api_repr(cls, api_repr):
+    def from_api_repr(cls, api_repr: dict) -> "TimePartitioning":
         """Return a :class:`TimePartitioning` object deserialized from a dict.
 
         This method creates a new ``TimePartitioning`` instance that points to
@@ -2178,7 +2188,7 @@ def from_api_repr(cls, api_repr):
         instance._properties = api_repr
         return instance
 
-    def to_api_repr(self):
+    def to_api_repr(self) -> dict:
         """Return a dictionary representing this object.
 
         This method returns the properties dict of the ``TimePartitioning``
diff --git a/noxfile.py b/noxfile.py
index 8523eabb5..f4e5f94fb 100644
--- a/noxfile.py
+++ b/noxfile.py
@@ -21,6 +21,7 @@
 import nox
 
 
+PYTYPE_VERSION = "pytype==2020.7.24"
 BLACK_VERSION = "black==19.10b0"
 BLACK_PATHS = ("docs", "google", "samples", "tests", "noxfile.py", "setup.py")
 CURRENT_DIRECTORY = pathlib.Path(__file__).parent.absolute()
@@ -86,6 +87,14 @@ def unit(session):
     default(session)
 
 
+@nox.session(python="3.7")
+def pytype(session):
+    """Run pytype
+    """
+    session.install(PYTYPE_VERSION)
+    session.run("pytype",)
+
+
 @nox.session(python=["3.8"])
 def system(session):
     """Run the system test suite."""
diff --git a/samples/geography/requirements.txt b/samples/geography/requirements.txt
index 3ea0e6e06..2dc755f9a 100644
--- a/samples/geography/requirements.txt
+++ b/samples/geography/requirements.txt
@@ -1,3 +1,4 @@
 geojson==2.5.0
 google-cloud-bigquery==2.6.1
+google-cloud-bigquery-storage==2.1.0
 Shapely==1.7.1
diff --git a/setup.cfg b/setup.cfg
index c3a2b39f6..0e2e0437b 100644
--- a/setup.cfg
+++ b/setup.cfg
@@ -17,3 +17,13 @@
 # Generated by synthtool. DO NOT EDIT!
 [bdist_wheel]
 universal = 1
+[pytype]
+python_version = 3.8
+inputs =
+    google/cloud/
+exclude =
+    tests/
+    google/cloud/bigquery_v2/
+output = .pytype/
+# Workaround for https://github.com/google/pytype/issues/150
+disable = pyi-error
\ No newline at end of file