Repository: googleapis/python-bigquery

Issue:
QueryJob class break flow after upgrade to 3.21.0
Affect version: >=3.21.0, <=3.23.1

#### Steps to reproduce

``` python
from google.cloud import bigquery
import uuid

client = bigquery.Client()
from google.cloud.bigquery import QueryJob
QueryJob(job_id=str(uuid.uuid4()), query="select 1", client=client).result() # return 404 error, work with version 3.20.1
```




Discussion:

Patch:
diff --git a/google/cloud/bigquery/job/query.py b/google/cloud/bigquery/job/query.py
index a8530271..8049b748 100644
--- a/google/cloud/bigquery/job/query.py
+++ b/google/cloud/bigquery/job/query.py
@@ -1585,6 +1585,11 @@ class QueryJob(_AsyncJob):
                     self._retry_do_query = retry_do_query
                     self._job_retry = job_retry
 
+                # If the job hasn't been created, create it now. Related:
+                # https://github.com/googleapis/python-bigquery/issues/1940
+                if self.state is None:
+                    self._begin(retry=retry, **done_kwargs)
+
                 # Refresh the job status with jobs.get because some of the
                 # exceptions thrown by jobs.getQueryResults like timeout and
                 # rateLimitExceeded errors are ambiguous. We want to know if


Test patch:
diff --git a/tests/unit/job/test_query.py b/tests/unit/job/test_query.py
index 66055dee..5b69c98c 100644
--- a/tests/unit/job/test_query.py
+++ b/tests/unit/job/test_query.py
@@ -1037,6 +1037,86 @@ class TestQueryJob(_Base):
         self.assertIsNone(result.job_id)
         self.assertIsNone(result.query_id)
 
+    # If the job doesn't exist, create the job first. Issue:
+    # https://github.com/googleapis/python-bigquery/issues/1940
+    def test_result_begin_job_if_not_exist(self):
+        begun_resource = self._make_resource()
+        query_running_resource = {
+            "jobComplete": True,
+            "jobReference": {
+                "projectId": self.PROJECT,
+                "jobId": self.JOB_ID,
+                "location": "US",
+            },
+            "schema": {"fields": [{"name": "col1", "type": "STRING"}]},
+            "status": {"state": "RUNNING"},
+        }
+        query_done_resource = {
+            "jobComplete": True,
+            "jobReference": {
+                "projectId": self.PROJECT,
+                "jobId": self.JOB_ID,
+                "location": "US",
+            },
+            "schema": {"fields": [{"name": "col1", "type": "STRING"}]},
+            "status": {"state": "DONE"},
+        }
+        done_resource = copy.deepcopy(begun_resource)
+        done_resource["status"] = {"state": "DONE"}
+        connection = make_connection(
+            begun_resource,
+            query_running_resource,
+            query_done_resource,
+            done_resource,
+        )
+        client = _make_client(project=self.PROJECT, connection=connection)
+        job = self._make_one(self.JOB_ID, self.QUERY, client)
+        job._properties["jobReference"]["location"] = "US"
+
+        job.result()
+
+        create_job_call = mock.call(
+            method="POST",
+            path=f"/projects/{self.PROJECT}/jobs",
+            data={
+                "jobReference": {
+                    "jobId": self.JOB_ID,
+                    "projectId": self.PROJECT,
+                    "location": "US",
+                },
+                "configuration": {
+                    "query": {"useLegacySql": False, "query": self.QUERY},
+                },
+            },
+            timeout=None,
+        )
+        reload_call = mock.call(
+            method="GET",
+            path=f"/projects/{self.PROJECT}/jobs/{self.JOB_ID}",
+            query_params={"projection": "full", "location": "US"},
+            timeout=DEFAULT_GET_JOB_TIMEOUT,
+        )
+        get_query_results_call = mock.call(
+            method="GET",
+            path=f"/projects/{self.PROJECT}/queries/{self.JOB_ID}",
+            query_params={
+                "maxResults": 0,
+                "location": "US",
+            },
+            timeout=None,
+        )
+
+        connection.api_request.assert_has_calls(
+            [
+                # Make sure we start a job that hasn't started yet. See:
+                # https://github.com/googleapis/python-bigquery/issues/1940
+                create_job_call,
+                reload_call,
+                get_query_results_call,
+                reload_call,
+            ]
+        )
+
     def test_result_with_done_job_calls_get_query_results(self):
         query_resource_done = {
             "jobComplete": True,
@@ -1379,6 +1459,7 @@ class TestQueryJob(_Base):
         client = _make_client(project=self.PROJECT, connection=connection)
         job = self._make_one(self.JOB_ID, self.QUERY, client)
         job._properties["jobReference"]["location"] = "US"
+        job._properties["status"] = {"state": "RUNNING"}
 
         with freezegun.freeze_time("1970-01-01 00:00:00", tick=False):
             job.result(
@@ -1429,6 +1510,7 @@ class TestQueryJob(_Base):
         client = _make_client(project=self.PROJECT, connection=connection)
         job = self._make_one(self.JOB_ID, self.QUERY, client)
         job._properties["jobReference"]["location"] = "US"
+        job._properties["status"] = {"state": "RUNNING"}
 
         with freezegun.freeze_time(
             "1970-01-01 00:00:00", auto_tick_seconds=1.0
@@ -2319,5 +2401,6 @@ class TestQueryJob(_Base):
         connection = make_connection(begun_resource, query_resource, done_resource)
         client = _make_client(project=self.PROJECT, connection=connection)
         job = self._make_one(self.JOB_ID, self.QUERY, client)
+        job._properties["status"] = {"state": "RUNNING"}
 
         self.assertIsInstance(iter(job), types.GeneratorType)