feat: added connection_properties and create_session in load job

.gitignore

@@ -51,6 +51,7 @@ docs.metadata
 
 # Virtual environment
 env/
+venv/
 
 # Test logs
 coverage.xml

google/cloud/bigquery/job/load.py

@@ -28,6 +28,7 @@
 from google.cloud.bigquery.job.base import _AsyncJob
 from google.cloud.bigquery.job.base import _JobConfig
 from google.cloud.bigquery.job.base import _JobReference
+from google.cloud.bigquery.query import ConnectionProperty
 
 
 class LoadJobConfig(_JobConfig):
@@ -120,6 +121,25 @@ def clustering_fields(self, value):
         else:
             self._del_sub_prop("clustering")
 
+    @property
+    def connection_properties(self) -> List[ConnectionProperty]:
+        """Connection properties.
+
+        See
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#JobConfigurationLoad.FIELDS.connection_properties
+
+        .. versionadded:: 3.7.0
+        """
+        resource = self._get_sub_prop("connectionProperties", [])
+        return [ConnectionProperty.from_api_repr(prop) for prop in resource]
+
+    @connection_properties.setter
+    def connection_properties(self, value: Iterable[ConnectionProperty]):
+        self._set_sub_prop(
+            "connectionProperties",
+            [prop.to_api_repr() for prop in value],
+        )
+
     @property
     def create_disposition(self):
         """Optional[google.cloud.bigquery.job.CreateDisposition]: Specifies behavior
@@ -134,6 +154,27 @@ def create_disposition(self):
     def create_disposition(self, value):
         self._set_sub_prop("createDisposition", value)
 
+    @property
+    def create_session(self) -> Optional[bool]:
+        """[Preview] If :data:`True`, creates a new session, where
+        :attr:`~google.cloud.bigquery.job.LoadJob.session_info` will contain a
+        random server generated session id.
+
+        If :data:`False`, runs load job with an existing ``session_id`` passed in
+        :attr:`~google.cloud.bigquery.job.LoadJobConfig.connection_properties`,
+        otherwise runs load job in non-session mode.
+
+        See
+        https://cloud.google.com/bigquery/docs/reference/rest/v2/Job#JobConfigurationLoad.FIELDS.create_session
+
+        .. versionadded:: 3.7.0
+        """
+        return self._get_sub_prop("createSession")
+
+    @create_session.setter
+    def create_session(self, value: Optional[bool]):
+        self._set_sub_prop("createSession", value)
+
     @property
     def decimal_target_types(self) -> Optional[FrozenSet[str]]:
         """Possible SQL data types to which the source decimal values are converted.
@@ -629,13 +670,31 @@ def autodetect(self):
         """
         return self._configuration.autodetect
 
+    @property
+    def connection_properties(self) -> List[ConnectionProperty]:
+        """See
+        :attr:`google.cloud.bigquery.job.LoadJobConfig.connection_properties`.
+
+        .. versionadded:: 3.7.0
+        """
+        return self._configuration.connection_properties
+
     @property
     def create_disposition(self):
         """See
         :attr:`google.cloud.bigquery.job.LoadJobConfig.create_disposition`.
         """
         return self._configuration.create_disposition
 
+    @property
+    def create_session(self) -> Optional[bool]:
+        """See
+        :attr:`google.cloud.bigquery.job.LoadJobConfig.create_session`.
+
+        .. versionadded:: 3.7.0
+        """
+        return self._configuration.create_session
+
     @property
     def encoding(self):
         """See

tests/unit/job/test_load.py

@@ -392,6 +392,8 @@ def test_from_api_repr_bare(self):
         job = klass.from_api_repr(RESOURCE, client=client)
         self.assertIs(job._client, client)
         self._verifyResourceProperties(job, RESOURCE)
+        self.assertEqual(len(job.connection_properties), 0)
+        self.assertIsNone(job.create_session)
 
     def test_from_api_with_encryption(self):
         self._setUpConstants()

tests/unit/job/test_load_config.py

@@ -122,6 +122,27 @@ def test_create_disposition_setter(self):
         config.create_disposition = disposition
         self.assertEqual(config._properties["load"]["createDisposition"], disposition)
 
+    def test_connection_properties(self):
+        from google.cloud.bigquery.query import ConnectionProperty
+
+        config = self._get_target_class()()
+        self.assertEqual(len(config.connection_properties), 0)
+
+        session_id = ConnectionProperty("session_id", "abcd")
+        time_zone = ConnectionProperty("time_zone", "America/Chicago")
+        config.connection_properties = [session_id, time_zone]
+        self.assertEqual(len(config.connection_properties), 2)
+        self.assertEqual(config.connection_properties[0].key, "session_id")
+        self.assertEqual(config.connection_properties[0].value, "abcd")
+        self.assertEqual(config.connection_properties[1].key, "time_zone")
+        self.assertEqual(config.connection_properties[1].value, "America/Chicago")
+
+    def test_create_session(self):
+        config = self._get_target_class()()
+        self.assertIsNone(config.create_session)
+        config.create_session = True
+        self.assertTrue(config.create_session)
+
     def test_decimal_target_types_miss(self):
         config = self._get_target_class()()
         self.assertIsNone(config.decimal_target_types)