Repository: simple-salesforce/simple-salesforce (language: python)

Issue:
Escaped password is causing authentication error
**lib ver:** 1.12.2
**api ver:** 52.0

Suppose you have `password&123`, it becomes `password&amp;123` in the payload that gets `POST`ed to the server.

I'm on version 52.0 and it's rejecting escaped passwords.  When I submit the `token_login` call without the escape, it works fine.

The  escapes are here at lines 83 in `simple_salesforce/login.py`:

```
# pylint: disable=E0012,deprecated-method
username = escape(username) if username else None
password = escape(password) if password else None
```


Discussion / hints:

Fix patch:
diff --git a/.pylintrc b/.pylintrc
index c8c2935e..df2259fe 100644
--- a/.pylintrc
+++ b/.pylintrc
@@ -38,7 +38,7 @@ load-plugins=
 # --enable=similarities". If you want to run only the classes checker, but have
 # no Warning level messages displayed, use"--disable=all --enable=classes
 # --disable=W"
-disable=too-many-lines,consider-using-f-string,too-many-public-methods,too-few-public-methods,no-member,too-many-ancestors,I0011,deprecated-disable-all,file-ignored,import-error,no-name-in-module,maybe-no-member,fixme,duplicate-code,cyclic-import,wrong-import-order,wrong-import-position,ungrouped-imports,len-as-condition
+disable=too-many-lines,too-many-public-methods,too-few-public-methods,no-member,too-many-ancestors,I0011,deprecated-disable-all,file-ignored,import-error,no-name-in-module,maybe-no-member,fixme,duplicate-code,cyclic-import,wrong-import-order,wrong-import-position,ungrouped-imports,len-as-condition,broad-exception-raised
 
 
 [REPORTS]
diff --git a/.travis.yml b/.travis.yml
index ada5afbf..ca18d5f2 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -5,6 +5,9 @@ python:
   - "3.6"
   - "3.7"
   - "3.8"
+  - "3.9"
+  - "3.10"
+  - "3.11"
 jobs:
   fast_finish: true
   include:
diff --git a/CHANGES b/CHANGES
index 4a428eb5..bc26d96f 100644
--- a/CHANGES
+++ b/CHANGES
@@ -1,3 +1,17 @@
+v1.12.3
+======
+Bug Fixes
+--------
+- [#594] Fix infinite refresh session issue
+- [#603] Fix upsert metadata
+- [#604] Fix escaped username and password in connected app
+
+Other
+--------
+- [#585] Additional documentation (README) on using package with Process Approvals rest API
+- [#597] Use f-strings
+- [#599] Use replace os.path with pathlib
+
 v1.12.2
 ======
 Features
diff --git a/README.rst b/README.rst
index fd974759..fe2dbeb9 100644
--- a/README.rst
+++ b/README.rst
@@ -9,7 +9,7 @@ Simple Salesforce
    :target: http://simple-salesforce.readthedocs.io/en/latest/?badge=latest
    :alt: Documentation Status
 
-Simple Salesforce is a basic Salesforce.com REST API client built for Python 3.6, 3.7 and 3.8. The goal is to provide a very low-level interface to the REST Resource and APEX API, returning a dictionary of the API JSON response.
+Simple Salesforce is a basic Salesforce.com REST API client built for Python 3.6, 3.7 3.8, 3.9, 3.10, and 3.11. The goal is to provide a very low-level interface to the REST Resource and APEX API, returning a dictionary of the API JSON response.
 
 =============
 
@@ -529,6 +529,121 @@ Hard deletion:
     sf.bulk.Contact.hard_delete(data,batch_size=10000,use_serial=True)
 
 
+Using Bulk 2.0
+--------------------------
+
+You can use this library to access Bulk 2.0 API functions.
+
+Create new records:
+
+.. code-block:: text
+
+    "Custom_Id__c","AccountId","Email","FirstName","LastName"
+    "CustomID1","ID-13","contact1@example.com","Bob","x"
+    "CustomID2","ID-24","contact2@example.com","Alice","y"
+    ...
+
+.. code-block:: python
+
+    sf.bulk2.Contact.insert("./sample.csv", batch_size=10000)
+
+
+Create new records concurrently:
+
+.. code-block:: python
+
+    sf.bulk2.Contact.insert("./sample.csv", batch_size=10000, concurrency=10)
+
+
+Update existing records:
+
+.. code-block:: text
+
+    "Custom_Id__c","AccountId","Email","FirstName","LastName"
+    "CustomID1","ID-13","contact1@example.com","Bob","X"
+    "CustomID2","ID-24","contact2@example.com","Alice","Y"
+    ...
+
+.. code-block:: python
+
+    sf.bulk2.Contact.update("./sample.csv")
+
+
+Upsert records:
+
+.. code-block:: text
+
+    "Custom_Id__c","LastName"
+    "CustomID1","X"
+    "CustomID2","Y"
+    ...
+
+.. code-block:: python
+
+    sf.bulk2.Contact.upsert("./sample.csv", 'Custom_Id__c')
+
+
+Query records:
+
+.. code-block:: python
+
+    query = 'SELECT Id, Name FROM Account LIMIT 100000'
+
+    results = sf.bulk2.Account.query(
+        query, max_records=50000, column_delimiter="COMM", line_ending="LF"
+    )
+    for i, data in enumerate(results):
+        with open(f"results/part-{1}.csv", "w") as bos:
+            bos.write(data)
+
+
+Download records(low memory usage):
+
+.. code-block:: python
+
+    query = 'SELECT Id, Name FROM Account'
+
+    sf.bulk2.Account.download(
+        query, path="results/", max_records=200000
+    )
+
+
+Delete records (soft deletion):
+
+.. code-block:: text
+
+    "Id"
+    "0000000000AAAAA"
+    "0000000000BBBBB"
+    ...
+
+
+.. code-block:: python
+
+    sf.bulk2.Contact.delete("./sample.csv")
+
+
+Hard deletion:
+
+.. code-block:: python
+
+    sf.bulk2.Contact.hard_delete("./sample.csv")
+
+
+Retrieve failed/successful/unprocessed records for ingest(insert,update...) job:
+
+.. code-block:: python
+
+    results = sf.bulk2.Contact.insert("./sample.csv")
+    # [{"numberRecordsFailed": 123, "numberRecordsProcessed": 2000, "numberRecordsTotal": 2000, "job_id": "Job-1"}, ...]
+    for result in results:
+        job_id = result['job_id']
+        # also available: get_unprocessed_records, get_successful_records
+        data = sf.bulk2.Contact.get_failed_records(job_id)
+        # or save to file
+        sf.bulk2.Contact.get_failed_records(job_id, file=f'{job_id}.csv')
+
+
 Using Apex
 --------------------------
 
diff --git a/docs/user_guide/other_REST_examples.rst b/docs/user_guide/other_REST_examples.rst
new file mode 100644
index 00000000..b70c0702
--- /dev/null
+++ b/docs/user_guide/other_REST_examples.rst
@@ -0,0 +1,72 @@
+Using REST
+----------
+
+In addition to built in functions that allow integrations with the Salesforce Apex and SOAP endpoints the Simple Salesforce package can be used to target the REST API. This example will demonstrate how to "Submit", `SFDC Submit documentation`_, a process approval in SFDC through the REST API.
+
+.. _SFDC Submit documentation: https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/dome_process_approvals_submit.htm
+
+1. Import any dependant packages to your script. For this example we use Simple Salesforce, Requests and the built in JSON package:
+
+.. code-block:: python
+
+    import json, requests, simple_salesforce
+
+
+1. Authenticate and create an instance of your Salesforce environment using SalesForceLogin:
+
+.. code-block:: python
+
+    session_id, instance = simple_salesforce.SalesforceLogin(
+            username=sf_user,
+            password=sf_password,
+            security_token=sf_token)
+
+2. Build the body of the request using information specific to your environment:
+
+.. code-block:: python
+
+    body = {"requests" : [{
+            "actionType": "Submit",
+            "contextId": "001D000000I8mIm",
+            "nextApproverIds": ["005D00000015rY9"],
+            "comments":"this is a test",
+            "contextActorId": "005D00000015rZy",
+            "processDefinitionNameOrId" : "PTO_Request_Process",
+            "skipEntryCriteria": "true"}]
+            }
+
+3. Generate the endpoint url and request headers dynamically using your specific SalesForceLogin instance and session id from above:
+
+.. code-block:: python
+
+    url = f"https://{instance}/services/data/v55.0/process/approvals/"
+    headers = {"Content-Type": "application/json", 'Authorization': 'Bearer ' + session_id}
+
+4. Use your favorite libraries, in this example we use Requests and JSON, to send a HTTP request to the REST API:
+
+.. code-block:: python
+
+    response = requests.post(url, data=json.dumps(body), headers=headers)
+
+5. A response will be returned in a JSON string that can be loaded into your script using the following:
+
+.. code-block:: python
+
+    resp = response.json()
+
+This response will match the following format:
+
+.. code-block:: python
+
+    [ {
+      "actorIds" : [ "005D00000015rY9IAI" ],
+       "entityId" : "001D000000I8mImIAJ",
+       "errors" : null,
+       "instanceId" : "04gD0000000Cvm5IAC",
+       "instanceStatus" : "Pending",
+       "newWorkitemIds" : [ "04iD0000000Cw6SIAS" ],
+       "success" : true } ]
+
+You can learn more about SFDC REST and the available calls on `Salesforce REST API Developer Guide`_
+
+.. _Salesforce REST API Developer Guide: https://developer.salesforce.com/docs/atlas.en-us.api_rest.meta/api_rest/intro_rest.htm
diff --git a/setup.py b/setup.py
index b22d2e93..66f79da8 100644
--- a/setup.py
+++ b/setup.py
@@ -1,16 +1,14 @@
 """Simple-Salesforce Package Setup"""
 
-import os
-import sys
 import textwrap
+from pathlib import Path
 
 from setuptools import setup
 
-here = os.path.abspath(os.path.dirname(__file__))
+here = Path(__file__).parent
 
 about = {}
-with open(os.path.join(here, 'simple_salesforce', '__version__.py'), 'r') as f:
-    exec(f.read(), about)
+exec((here / 'simple_salesforce' / '__version__.py').read_text(), about)
 
 setup(
     name=about['__title__'],
@@ -23,23 +21,26 @@
     url=about['__url__'],
     license=about['__license__'],
     description=about['__description__'],
-    long_description=textwrap.dedent(open('README.rst', 'r').read()),
+    long_description=textwrap.dedent((here / 'README.rst').read_text()),
     long_description_content_type='text/x-rst',
     package_data={
         'simple_salesforce': ['metadata.wsdl'],
         },
 install_requires = [
                        'requests>=2.22.0',
-                       'authlib',
-                       'zeep'
+                       'cryptography',
+                       'zeep',
+                       'pyjwt',
+                       'more-itertools',
+                       'pendulum'
                        ],
                    tests_require = [
-                                       'nose>=1.3.0',
+                                       'pytest',
                                        'pytz>=2014.1.1',
                                        'responses>=0.5.1',
                                        'cryptography<3.4',
                                        ],
-                                   test_suite = 'nose.collector',
+                                   test_suite = 'simple_salesforce.tests',
                                                 keywords = about[
                                                                '__keywords__'],
                                                            classifiers = [
@@ -52,6 +53,9 @@
     'Programming Language :: Python :: 3.6',
     'Programming Language :: Python :: 3.7',
     'Programming Language :: Python :: 3.8',
+    'Programming Language :: Python :: 3.9',
+    'Programming Language :: Python :: 3.10',
+    'Programming Language :: Python :: 3.11',
     'Programming Language :: Python :: Implementation :: PyPy'
     ]
 )
diff --git a/simple_salesforce/__version__.py b/simple_salesforce/__version__.py
index fb115f26..7280d720 100644
--- a/simple_salesforce/__version__.py
+++ b/simple_salesforce/__version__.py
@@ -5,7 +5,7 @@
 __title__ = 'simple-salesforce'
 __description__ = 'A basic Salesforce.com REST API client.'
 __url__ = 'https://github.com/simple-salesforce/simple-salesforce'
-__version__ = '1.12.2'
+__version__ = '1.12.3'
 __author__ = 'Nick Catalano'
 __author_email__ = 'nickcatal@gmail.com'
 __license__ = 'Apache 2.0'
diff --git a/simple_salesforce/api.py b/simple_salesforce/api.py
index ee40ac84..e4a9e37a 100644
--- a/simple_salesforce/api.py
+++ b/simple_salesforce/api.py
@@ -8,11 +8,13 @@
 import re
 from collections import OrderedDict, namedtuple
 from functools import partial
+from pathlib import Path
 from urllib.parse import urljoin, urlparse
 
 import requests
 
 from .bulk import SFBulkHandler
+from .bulk2 import SFBulk2Handler
 from .exceptions import SalesforceGeneralError
 from .login import SalesforceLogin
 from .metadata import SfdcMetadataApi
@@ -156,7 +158,7 @@ def __init__(
                 self.sf_instance = urlparse(instance_url).hostname
                 port = urlparse(instance_url).port
                 if port not in (None, 443):
-                    self.sf_instance += ':' + str(port)
+                    self.sf_instance += f':{port}'
             else:
                 self.sf_instance = instance
 
@@ -181,7 +183,7 @@ def __init__(
             self._refresh_session()
 
         elif all(arg is not None for arg in (
-                username, consumer_key, consumer_secret)):
+                username, password, consumer_key, consumer_secret)):
             self.auth_type = "password"
 
             # Pass along the username/password to our login helper
@@ -189,6 +191,7 @@ def __init__(
                 SalesforceLogin,
                 session=self.session,
                 username=username,
+                password=password,
                 consumer_key=consumer_key,
                 consumer_secret=consumer_secret,
                 proxies=self.proxies,
@@ -216,8 +219,7 @@ def __init__(
                 'You must provide login information or an instance and token'
                 )
 
-        self.auth_site = ('https://{domain}.salesforce.com'
-                          .format(domain=self.domain))
+        self.auth_site = f'https://{self.domain}.salesforce.com'
 
         self.headers = {
             'Content-Type': 'application/json',
@@ -225,20 +227,18 @@ def __init__(
             'X-PrettyPrint': '1'
             }
 
-        self.base_url = ('https://{instance}/services/data/v{version}/'
-                         .format(instance=self.sf_instance,
-                                 version=self.sf_version))
-        self.apex_url = ('https://{instance}/services/apexrest/'
-                         .format(instance=self.sf_instance))
-        self.bulk_url = ('https://{instance}/services/async/{version}/'
-                         .format(instance=self.sf_instance,
-                                 version=self.sf_version))
-        self.metadata_url = ('https://{instance}/services/Soap/m/{version}/'
-                             .format(instance=self.sf_instance,
-                                     version=self.sf_version))
-        self.tooling_url = '{base_url}tooling/'.format(base_url=self.base_url)
-        self.oauth2_url = ('https://{instance}/services/oauth2/'
-                           .format(instance=self.sf_instance))
+        self.base_url = (
+            f'https://{self.sf_instance}/services/data/v{self.sf_version}/')
+        self.apex_url = f'https://{self.sf_instance}/services/apexrest/'
+        self.bulk_url = (
+            f'https://{self.sf_instance}/services/async/{self.sf_version}/')
+        self.bulk2_url = (
+            f'https://{self.sf_instance}/services/data/v{self.sf_version}/jobs/'
+        )
+        self.metadata_url = (
+            f'https://{self.sf_instance}/services/Soap/m/{self.sf_version}/')
+        self.tooling_url = f'{self.base_url}tooling/'
+        self.oauth2_url = f'https://{self.sf_instance}/services/oauth2/'
         self.api_usage = {}
         self._parse_float = parse_float
         self._object_pairs_hook = object_pairs_hook
@@ -320,6 +320,9 @@ def __getattr__(self, name):
             # Deal with bulk API functions
             return SFBulkHandler(self.session_id, self.bulk_url, self.proxies,
                                  self.session)
+        if name == 'bulk2':
+            return SFBulk2Handler(self.session_id, self.bulk2_url, self.proxies,
+                                  self.session)
 
         return SFType(
             name, self.session_id, self.sf_instance, sf_version=self.sf_version,
@@ -336,7 +339,7 @@ def set_password(self, user, password):
         * password: the new password
         """
 
-        url = self.base_url + 'sobjects/User/%s/password' % user
+        url = f'{self.base_url}sobjects/User/{user}/password'
         params = {'NewPassword': password}
 
         result = self._call_salesforce('POST', url, data=json.dumps(params))
@@ -424,7 +427,7 @@ def quick_search(self, search):
                     string will be wrapped to read `FIND {Waldo}` before being
                     sent to Salesforce
         """
-        search_string = 'FIND {{{search_string}}}'.format(search_string=search)
+        search_string = f'FIND {{{search}}}'
         return self.search(search_string)
 
     def limits(self, **kwargs):
@@ -476,14 +479,10 @@ def query_more(
         """
         if identifier_is_url:
             # Don't use `self.base_url` here because the full URI is provided
-            url = ('https://{instance}{next_record_url}'
-                   .format(instance=self.sf_instance,
-                           next_record_url=next_records_identifier))
+            url = f'https://{self.sf_instance}{next_records_identifier}'
         else:
             endpoint = 'queryAll' if include_deleted else 'query'
-            url = self.base_url + '{query_endpoint}/{next_record_id}'
-            url = url.format(query_endpoint=endpoint,
-                             next_record_id=next_records_identifier)
+            url = f'{self.base_url}{endpoint}/{next_records_identifier}'
         result = self._call_salesforce('GET', url, name='query_more', **kwargs)
 
         return self.parse_result_to_json(result)
@@ -506,8 +505,7 @@ def query_all_iter(self, query, include_deleted=False, **kwargs):
 
         result = self.query(query, include_deleted=include_deleted, **kwargs)
         while True:
-            for record in result['records']:
-                yield record
+            yield from result['records']
             # fetch next batch if we're not done else break out of loop
             if not result['done']:
                 result = self.query_more(result['nextRecordsUrl'],
@@ -602,8 +600,10 @@ def _call_salesforce(self, method, url, name="", **kwargs):
 
         if self._salesforce_login_partial is not None \
                 and result.status_code == 401:
-            self._refresh_session()
-            return self._call_salesforce(method, url, name, **kwargs)
+            error_details = result.json()[0]
+            if error_details['errorCode'] == 'INVALID_SESSION_ID':
+                self._refresh_session()
+                return self._call_salesforce(method, url, name, **kwargs)
 
         if result.status_code >= 300:
             exception_handler(result, name=name)
@@ -742,10 +742,8 @@ def __init__(
         self.api_usage = {}
 
         self.base_url = (
-            'https://{instance}/services/data/v{sf_version}/sobjects'
-            '/{object_name}/'.format(instance=sf_instance,
-                                     object_name=object_name,
-                                     sf_version=sf_version))
+            f'https://{sf_instance}/services/data/v{sf_version}/sobjects'
+            f'/{object_name}/')
 
     @property
     def session_id(self):
@@ -784,9 +782,7 @@ def describe_layout(self, record_id, headers=None):
         * record_id -- the Id of the SObject to get
         * headers -- a dict with additional request headers.
         """
-        custom_url_part = 'describe/layouts/{record_id}'.format(
-            record_id=record_id
-            )
+        custom_url_part = f'describe/layouts/{record_id}'
         result = self._call_salesforce(
             method='GET',
             url=urljoin(self.base_url, custom_url_part),
@@ -818,11 +814,7 @@ def get_by_custom_id(self, custom_id_field, custom_id, headers=None):
         * custom_id - the External ID value of the SObject to get
         * headers -- a dict with additional request headers.
         """
-        custom_url = urljoin(
-            self.base_url, '{custom_id_field}/{custom_id}'.format(
-                custom_id_field=custom_id_field, custom_id=custom_id
-                )
-            )
+        custom_url = urljoin(self.base_url, f'{custom_id_field}/{custom_id}')
         result = self._call_salesforce(
             method='GET', url=custom_url, headers=headers
             )
@@ -913,9 +905,8 @@ def deleted(self, start, end, headers=None):
         * headers -- a dict with additional request headers.
         """
         url = urljoin(
-            self.base_url, 'deleted/?start={start}&end={end}'.format(
-                start=date_to_iso8601(start), end=date_to_iso8601(end)
-                )
+            self.base_url,
+            f'deleted/?start={date_to_iso8601(start)}&end={date_to_iso8601(end)}'
             )
         result = self._call_salesforce(method='GET', url=url, headers=headers)
         return self.parse_result_to_json(result)
@@ -932,9 +923,8 @@ def updated(self, start, end, headers=None):
         * headers -- a dict with additional request headers.
         """
         url = urljoin(
-            self.base_url, 'updated/?start={start}&end={end}'.format(
-                start=date_to_iso8601(start), end=date_to_iso8601(end)
-                )
+            self.base_url,
+            f'updated/?start={date_to_iso8601(start)}&end={date_to_iso8601(end)}'
             )
         result = self._call_salesforce(method='GET', url=url, headers=headers)
         return self.parse_result_to_json(result)
@@ -956,8 +946,10 @@ def _call_salesforce(self, method, url, **kwargs):
         if (self.salesforce
                 and self.salesforce._salesforce_login_partial is not None
                 and result.status_code == 401):
-            self.salesforce._refresh_session()
-            return self._call_salesforce(method, url, **kwargs)
+            error_details = result.json()[0]
+            if error_details['errorCode'] == 'INVALID_SESSION_ID':
+                self.salesforce._refresh_session()
+                return self._call_salesforce(method, url, **kwargs)
 
         if result.status_code >= 300:
             exception_handler(result, self.name)
@@ -988,8 +980,7 @@ def upload_base64(self, file_path, base64_field='Body', headers=None,
                       **kwargs):
         """Upload base64 encoded file to Salesforce"""
         data = {}
-        with open(file_path, "rb") as f:
-            body = base64.b64encode(f.read()).decode('utf-8')
+        body = base64.b64encode(Path(file_path).read_bytes()).decode()
         data[base64_field] = body
         result = self._call_salesforce(method='POST', url=self.base_url,
                                        headers=headers, json=data, **kwargs)
@@ -1001,8 +992,7 @@ def update_base64(self, record_id, file_path, base64_field='Body',
                       **kwargs):
         """Updated base64 image from file to Salesforce"""
         data = {}
-        with open(file_path, "rb") as f:
-            body = base64.b64encode(f.read()).decode('utf-8')
+        body = base64.b64encode(Path(file_path).read_bytes()).decode()
         data[base64_field] = body
         result = self._call_salesforce(method='PATCH',
                                        url=urljoin(self.base_url, record_id),
@@ -1022,8 +1012,7 @@ def get_base64(self, record_id, base64_field='Body', data=None,
                      sobjects/ContentVersion/ABC123/VersionData
         """
         result = self._call_salesforce(method='GET', url=urljoin(
-            self.base_url, '{record_id}/{base64_field}'.format(
-                record_id=record_id, base64_field=base64_field)),
+            self.base_url, f'{record_id}/{base64_field}'),
                                        data=data,
                                        headers=headers, **kwargs)
 
diff --git a/simple_salesforce/bulk.py b/simple_salesforce/bulk.py
index a232a781..6df18aee 100644
--- a/simple_salesforce/bulk.py
+++ b/simple_salesforce/bulk.py
@@ -98,7 +98,7 @@ def _create_job(self, operation, use_serial,
         if operation == 'upsert':
             payload['externalIdFieldName'] = external_id_field
 
-        url = "{}{}".format(self.bulk_url, 'job')
+        url = f'{self.bulk_url}job'
 
         result = call_salesforce(url=url, method='POST', session=self.session,
                                  headers=self.headers,
@@ -111,7 +111,7 @@ def _close_job(self, job_id):
             'state': 'Closed'
             }
 
-        url = "{}{}{}".format(self.bulk_url, 'job/', job_id)
+        url = f'{self.bulk_url}job/{job_id}'
 
         result = call_salesforce(url=url, method='POST', session=self.session,
                                  headers=self.headers,
@@ -120,7 +120,7 @@ def _close_job(self, job_id):
 
     def _get_job(self, job_id):
         """ Get an existing job to check the status """
-        url = "{}{}{}".format(self.bulk_url, 'job/', job_id)
+        url = f'{self.bulk_url}job/{job_id}'
 
         result = call_salesforce(url=url, method='GET', session=self.session,
                                  headers=self.headers)
@@ -132,7 +132,7 @@ def _add_batch(self, job_id, data, operation):
         implementations involving multiple batches
         """
 
-        url = "{}{}{}{}".format(self.bulk_url, 'job/', job_id, '/batch')
+        url = f'{self.bulk_url}job/{job_id}/batch'
 
         if operation not in ('query', 'queryAll'):
             data = json.dumps(data, allow_nan=False)
@@ -144,8 +144,7 @@ def _add_batch(self, job_id, data, operation):
     def _get_batch(self, job_id, batch_id):
         """ Get an existing batch to check the status """
 
-        url = "{}{}{}{}{}".format(self.bulk_url, 'job/',
-                                  job_id, '/batch/', batch_id)
+        url = f'{self.bulk_url}job/{job_id}/batch/{batch_id}'
 
         result = call_salesforce(url=url, method='GET', session=self.session,
                                  headers=self.headers)
@@ -154,15 +153,14 @@ def _get_batch(self, job_id, batch_id):
     def _get_batch_results(self, job_id, batch_id, operation):
         """ retrieve a set of results from a completed job """
 
-        url = "{}{}{}{}{}{}".format(self.bulk_url, 'job/', job_id, '/batch/',
-                                    batch_id, '/result')
+        url = f'{self.bulk_url}job/{job_id}/batch/{batch_id}/result'
 
         result = call_salesforce(url=url, method='GET', session=self.session,
                                  headers=self.headers)
 
         if operation in ('query', 'queryAll'):
             for batch_result in result.json():
-                url_query_results = "{}{}{}".format(url, '/', batch_result)
+                url_query_results = f'{url}/{batch_result}'
                 batch_query_result = call_salesforce(url=url_query_results,
                                                      method='GET',
                                                      session=self.session,
@@ -209,10 +207,12 @@ def _add_autosized_batches(self, data, operation, job):
         AND
         number_of_character_limit <= 10000000
 
-        testing for number of characters ensures that file size limit is
-        respected.
+        Documentation on limits can be found at:
+        https://developer.salesforce.com/docs/atlas.en-us.salesforce_app_limits_cheatsheet.meta/salesforce_app_limits_cheatsheet/salesforce_app_limits_platform_bulkapi.htm#ingest_jobs
 
-        this is due to json serialization of multibyte characters.
+        Our JSON serialization uses the default `ensure_ascii=True`, so the
+        character and byte lengths will be the same. Therefore we only need
+        to adhere to a single length limit of 10,000,000 characters.
 
         TODO: In future when simple-salesforce supports bulk api V2
         we should detect api version and set max file size accordingly. V2
@@ -224,28 +224,28 @@ def _add_autosized_batches(self, data, operation, job):
         * Maximum number of characters in a record: 400,000
         * Maximum number of characters in a field: 131,072
         """
-        file_limit = 1024 * 1024 * 10  # 10MB in bytes
-        rec_limit = 10000
-        char_limit = 10000000
+        record_limit = 10_000
+        char_limit = 10_000_000
 
         batches = []
         last_break = 0
-        nrecs, recsize = 0, 0
-        for i, rec in enumerate(data):
-            # 2 is added to account for the enclosing `[]`
-            # and the separator `, ` between records.
-            recsize += len(json.dumps(rec, default=str)) + 2
-            nrecs += len(rec)
+        record_count, char_count = 0, 0
+        for i, record in enumerate(data):
+            # 2 is added to account for the enclosing `[]` for the first record
+            # and the separator `, ` between records for subsequent records.
+            additional_chars = len(json.dumps(record, default=str)) + 2
             if any([
-                recsize >= file_limit,
-                recsize >= char_limit,
-                nrecs >= rec_limit,
-                i >= len(data) - 1
-                ]):
-                i = 1 if len(data) == 1 else i
+                char_count + additional_chars > char_limit,
+                record_count == record_limit
+            ]):
                 batches.append(data[last_break:i])
                 last_break = i
-                nrecs, recsize = 0, 0
+                record_count, char_count = 0, 0
+            char_count += additional_chars
+            record_count += 1
+        if last_break < len(data) - 1:
+            batches.append(data[last_break:])
+
         return [self._add_batch(job_id=job, data=i,
                                 operation=operation) for i in batches]
 
@@ -270,6 +270,10 @@ def _bulk_operation(self, operation, data, use_serial=False,
             raise ValueError('batch size should be auto or an integer')
 
         if operation not in ('query', 'queryAll'):
+            # Checks if data is present
+            if not data:
+                raise ValueError(f'data should not be empty for {operation}')
+
             # Checks to prevent batch limit
             if batch_size != 'auto':
                 batch_size = min(batch_size, len(data), 10000)
@@ -289,7 +293,7 @@ def _bulk_operation(self, operation, data, use_serial=False,
                                         operation=operation)
                         for i in
                         [data[i * batch_size:(i + 1) * batch_size]
-                        for i in range((len(data) // batch_size + 1))] if i]
+                        for i in range(len(data) // batch_size + 1)] if i]
 
                 multi_thread_worker = partial(self.worker,
                                               operation=operation,
diff --git a/simple_salesforce/bulk2.py b/simple_salesforce/bulk2.py
new file mode 100644
index 00000000..0d4d0820
--- /dev/null
+++ b/simple_salesforce/bulk2.py
@@ -0,0 +1,853 @@
+""" Classes for interacting with Salesforce Bulk 2.0 API """
+
+import copy
+import http.client as http
+import json
+import os
+import re
+import tempfile
+from collections import OrderedDict
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import closing
+from enum import Enum
+from functools import partial
+from time import sleep
+from typing import Dict, Tuple, Union, Generator, List
+
+import math
+import pendulum
+import requests
+from more_itertools import chunked
+from pendulum import DateTime
+
+from .exceptions import (
+    SalesforceBulkV2ExtractError,
+    SalesforceBulkV2LoadError,
+    SalesforceOperationError,
+)
+from .util import call_salesforce
+
+
+# pylint: disable=missing-class-docstring,invalid-name,too-many-arguments,too-many-locals
+
+
+class Operation(str, Enum):
+    insert = "insert"
+    upsert = "upsert"
+    update = "update"
+    delete = "delete"
+    hard_delete = "hardDelete"
+    query = "query"
+    query_all = "queryAll"
+
+
+class JobState(str, Enum):
+    open = "Open"
+    aborted = "Aborted"
+    failed = "Failed"
+    upload_complete = "UploadComplete"
+    in_progress = "InProgress"
+    job_complete = "JobComplete"
+
+
+class ColumnDelimiter(str, Enum):
+    BACKQUOTE = "BACKQUOTE"  # (`)
+    CARET = "CARET"  # (^)
+    COMMA = "COMMA"  # (,)
+    PIPE = "PIPE"  # (|)
+    SEMICOLON = "SEMICOLON"  # (;)
+    TAB = "TAB"  # (\t)
+
+
+_delimiter_char = {
+    ColumnDelimiter.BACKQUOTE: "`",
+    ColumnDelimiter.CARET: "^",
+    ColumnDelimiter.COMMA: ",",
+    ColumnDelimiter.PIPE: "|",
+    ColumnDelimiter.SEMICOLON: ";",
+    ColumnDelimiter.TAB: "\t",
+}
+
+
+class LineEnding(str, Enum):
+    LF = "LF"
+    CRLF = "CRLF"
+
+
+_line_ending_char = {LineEnding.LF: "\n", LineEnding.CRLF: "\r\n"}
+
+
+class ResultsType(str, Enum):
+    failed = "failedResults"
+    successful = "successfulResults"
+    unprocessed = "unprocessedRecords"
+
+
+# https://developer.salesforce.com/docs/atlas.en-us.242.0.salesforce_app_limits_cheatsheet.meta/salesforce_app_limits_cheatsheet/salesforce_app_limits_platform_bulkapi.htm
+# https://developer.salesforce.com/docs/atlas.en-us.api_asynch.meta/api_asynch/datafiles_prepare_csv.htm
+MAX_INGEST_JOB_FILE_SIZE = 100 * 1024 * 1024
+MAX_INGEST_JOB_PARALLELISM = 10  # TODO: ? Salesforce limits
+DEFAULT_QUERY_PAGE_SIZE = 50000
+
+
+def _split_csv(filename, max_records: int = None):
+    """Split a CSV file into chunks to avoid exceeding the Salesforce
+    bulk 2.0 API limits.
+
+    Arguments:
+        * filename -- csv file
+        * max_records -- the number of records per chunk, None for auto size
+    """
+    total_records = _count_csv(filename, skip_header=True)
+    max_records = max_records or total_records
+    max_records = min(max_records, total_records)
+    max_bytes = min(
+        os.path.getsize(filename), MAX_INGEST_JOB_FILE_SIZE - 1 * 1024 * 1024
+    )  # -1 MB for sentinel
+    records_size = 0
+    bytes_size = 0
+    buff = []
+    with open(filename, encoding="utf-8") as bis:
+        header = bis.readline()
+        for line in bis:
+            records_size += 1
+            bytes_size += len(line.encode("utf-8"))
+            if records_size > max_records or bytes_size > max_bytes:
+                if buff:
+                    yield records_size - 1, header + "".join(buff)
+                buff = [line]
+                records_size = 1
+                bytes_size = len(line.encode("utf-8"))
+            else:
+                buff.append(line)
+        if buff:
+            yield records_size, header + "".join(buff)
+
+
+def _count_csv(
+    filename=None, data=None, skip_header=False, line_ending=LineEnding.LF
+):
+    """Count the number of records in a CSV file."""
+    if filename:
+        with open(filename, encoding="utf-8") as bis:
+            count = sum(1 for _ in bis)
+    elif data:
+        pat = repr(_line_ending_char[line_ending])[1:-1]
+        count = sum(1 for _ in re.finditer(pat, data))
+    else:
+        raise ValueError("Either filename or data must be provided")
+
+    if skip_header:
+        count -= 1
+    return count
+
+
+class SFBulk2Handler:
+    """Bulk 2.0 API request handler
+    Intermediate class which allows us to use commands,
+     such as 'sf.bulk2.Contacts.insert(...)'
+    This is really just a middle layer, whose sole purpose is
+    to allow the above syntax
+    """
+
+    def __init__(self, session_id, bulk2_url, proxies=None, session=None):
+        """Initialize the instance with the given parameters.
+
+        Arguments:
+
+        * session_id -- the session ID for authenticating to Salesforce
+        * bulk2_url -- 2.0 API endpoint set in Salesforce instance
+        * proxies -- the optional map of scheme to proxy server
+        * session -- Custom requests session, created in calling code. This
+                     enables the use of requests Session features not otherwise
+                     exposed by simple_salesforce.
+        """
+        self.session_id = session_id
+        self.session = session or requests.Session()
+        self.bulk2_url = bulk2_url
+        # don't wipe out original proxies with None
+        if not session and proxies is not None:
+            self.session.proxies = proxies
+
+        # Define these headers separate from Salesforce class,
+        # as bulk uses a slightly different format
+        self.headers = {
+            "Content-Type": "application/json",
+            "Authorization": "Bearer " + self.session_id,
+            "X-PrettyPrint": "1",
+        }
+
+    def __getattr__(self, name):
+        return SFBulk2Type(
+            object_name=name,
+            bulk2_url=self.bulk2_url,
+            headers=self.headers,
+            session=self.session,
+        )
+
+
+class _Bulk2Client:
+    """Bulk 2.0 API client"""
+
+    JSON_CONTENT_TYPE = "application/json"
+    CSV_CONTENT_TYPE = "text/csv"
+
+    DEFAULT_WAIT_TIMEOUT_SECONDS = 86400  # 24-hour bulk job running time
+    MAX_CHECK_INTERVAL_SECONDS = 2.0
+
+    def __init__(self, object_name, bulk2_url, headers, session):
+        """
+        Arguments:
+
+        * object_name -- the name of the type of SObject this represents,
+                         e.g. `Lead` or `Contact`
+        * bulk2_url -- 2.0 API endpoint set in Salesforce instance
+        * headers -- bulk 2.0 API headers
+        * session -- Custom requests session, created in calling code. This
+                     enables the use of requests Session features not otherwise
+                     exposed by simple_salesforce.
+        """
+        self.object_name = object_name
+        self.bulk2_url = bulk2_url
+        self.session = session
+        self.headers = headers
+
+    def _get_headers(self, request_content_type=None, accept_content_type=None):
+        """Get headers for bulk 2.0 API request"""
+        headers = copy.deepcopy(self.headers)
+        headers["Content-Type"] = request_content_type or self.JSON_CONTENT_TYPE
+        headers["ACCEPT"] = accept_content_type or self.JSON_CONTENT_TYPE
+        return headers
+
+    def _construct_request_url(self, job_id, is_query: bool):
+        """Construct bulk 2.0 API request URL"""
+        if not job_id:
+            job_id = ""
+        if is_query:
+            url = self.bulk2_url + "query"
+        else:
+            url = self.bulk2_url + "ingest"
+        if job_id:
+            url = f"{url}/{job_id}"
+        return url
+
+    def create_job(
+        self,
+        operation,
+        query=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        external_id_field=None,
+    ):
+        """Create job
+
+        Arguments:
+
+        * operation -- Bulk operation to be performed by job
+        * query -- SOQL query to be performed by job
+        * column_delimiter -- The column delimiter used for CSV job data
+        * line_ending -- The line ending used for CSV job data
+        * external_id_field -- The external ID field in the object being updated
+        """
+        payload = {
+            "operation": operation,
+            "columnDelimiter": column_delimiter,
+            "lineEnding": line_ending,
+        }
+        if external_id_field:
+            payload["externalIdFieldName"] = external_id_field
+
+        is_query = operation in (Operation.query, Operation.query_all)
+        url = self._construct_request_url(None, is_query)
+        if is_query:
+            headers = self._get_headers(
+                self.JSON_CONTENT_TYPE, self.CSV_CONTENT_TYPE
+            )
+            if not query:
+                raise SalesforceBulkV2ExtractError(
+                    "Query is required for query jobs"
+                )
+            payload["query"] = query
+        else:
+            headers = self._get_headers(
+                self.JSON_CONTENT_TYPE, self.JSON_CONTENT_TYPE
+            )
+            payload["object"] = self.object_name
+            payload["contentType"] = "CSV"
+        result = call_salesforce(
+            url=url,
+            method="POST",
+            session=self.session,
+            headers=headers,
+            data=json.dumps(payload, allow_nan=False),
+        )
+        return result.json(object_pairs_hook=OrderedDict)
+
+    def wait_for_job(self, job_id, is_query: bool, wait=0.5):
+        """Wait for job completion or timeout"""
+        expiration_time: DateTime = pendulum.now().add(
+            seconds=self.DEFAULT_WAIT_TIMEOUT_SECONDS
+        )
+        job_status = JobState.in_progress if is_query else JobState.open
+        delay_timeout = 0.0
+        delay_cnt = 0
+        sleep(wait)
+        while pendulum.now() < expiration_time:
+            job_info = self.get_job(job_id, is_query)
+            job_status = job_info["state"]
+            if job_status in [
+                JobState.job_complete,
+                JobState.aborted,
+                JobState.failed,
+            ]:
+                if job_status != JobState.job_complete:
+                    error_message = job_info.get("errorMessage") or job_info
+                    raise SalesforceOperationError(
+                        f"Job failure. Response content: {error_message}"
+                    )
+                return job_status  # JobComplete
+
+            if delay_timeout < self.MAX_CHECK_INTERVAL_SECONDS:
+                delay_timeout = wait + math.exp(delay_cnt) / 1000.0
+                delay_cnt += 1
+            sleep(delay_timeout)
+        raise SalesforceOperationError(f"Job timeout. Job status: {job_status}")
+
+    def abort_job(self, job_id, is_query: bool):
+        """Abort query/ingest job"""
+        return self._set_job_state(job_id, is_query, JobState.aborted)
+
+    def close_job(self, job_id):
+        """Close ingest job"""
+        return self._set_job_state(job_id, False, JobState.upload_complete)
+
+    def delete_job(self, job_id, is_query: bool):
+        """Delete query/ingest job"""
+        url = self._construct_request_url(job_id, is_query)
+        headers = self._get_headers()
+        result = call_salesforce(
+            url=url, method="DELETE", session=self.session, headers=headers
+        )
+        return result.json(object_pairs_hook=OrderedDict)
+
+    def _set_job_state(self, job_id, is_query: bool, state: str):
+        """Set job state"""
+        url = self._construct_request_url(job_id, is_query)
+        headers = self._get_headers()
+        payload = {"state": state}
+        result = call_salesforce(
+            url=url,
+            method="PATCH",
+            session=self.session,
+            headers=headers,
+            data=json.dumps(payload, allow_nan=False),
+        )
+        return result.json(object_pairs_hook=OrderedDict)
+
+    def get_job(self, job_id, is_query: bool):
+        """Get job info"""
+        url = self._construct_request_url(job_id, is_query)
+
+        result = call_salesforce(
+            url=url, method="GET", session=self.session, headers=self.headers
+        )
+        return result.json(object_pairs_hook=OrderedDict)
+
+    def filter_null_bytes(self, b: Union[str, bytes]):
+        """
+        https://github.com/airbytehq/airbyte/issues/8300
+        """
+        if isinstance(b, str):
+            return b.replace("\x00", "")
+        if isinstance(b, bytes):
+            return b.replace(b"\x00", b"")
+        raise TypeError("Expected str or bytes")
+
+    def get_query_results(
+        self, job_id, locator: str = "", max_records=DEFAULT_QUERY_PAGE_SIZE
+    ):
+        """Get results for a query job"""
+        url = self._construct_request_url(job_id, True) + "/results"
+        params = {"maxRecords": max_records}
+        if locator and locator != "null":
+            params["locator"] = locator
+        headers = self._get_headers(
+            self.JSON_CONTENT_TYPE, self.CSV_CONTENT_TYPE
+        )
+        result = call_salesforce(
+            url=url,
+            method="GET",
+            session=self.session,
+            headers=headers,
+            params=params,
+        )
+        locator = result.headers.get("Sforce-Locator", "")
+        if locator == "null":
+            locator = ""
+        number_of_records = int(result.headers.get("Sforce-NumberOfRecords"))
+        return {
+            "locator": locator,
+            "number_of_records": number_of_records,
+            "records": self.filter_null_bytes(result.text),
+        }
+
+    def download_job_data(
+        self,
+        path,
+        job_id,
+        locator: str = "",
+        max_records=DEFAULT_QUERY_PAGE_SIZE,
+        chunk_size=1024,
+    ):
+        """Get results for a query job"""
+        if not os.path.exists(path):
+            raise SalesforceBulkV2LoadError(f"Path does not exist: {path}")
+
+        url = self._construct_request_url(job_id, True) + "/results"
+        params = {"maxRecords": max_records}
+        if locator and locator != "null":
+            params["locator"] = locator
+        headers = self._get_headers(
+            self.JSON_CONTENT_TYPE, self.CSV_CONTENT_TYPE
+        )
+        with closing(
+            call_salesforce(
+                url=url,
+                method="GET",
+                session=self.session,
+                headers=headers,
+                params=params,
+                stream=True,
+            )
+        ) as result, tempfile.NamedTemporaryFile(
+            "wb", dir=path, suffix=".csv", delete=False
+        ) as bos:
+            locator = result.headers.get("Sforce-Locator", "")
+            if locator == "null":
+                locator = ""
+            number_of_records = int(
+                result.headers.get("Sforce-NumberOfRecords")
+            )
+            for chunk in result.iter_content(chunk_size=chunk_size):
+                bos.write(self.filter_null_bytes(chunk))
+            # check the file exists
+            if os.path.isfile(bos.name):
+                return {
+                    "locator": locator,
+                    "number_of_records": number_of_records,
+                    "file": bos.name,
+                }
+            raise SalesforceBulkV2LoadError(
+                f"The IO/Error occured while verifying binary data. "
+                f"File {bos.name} doesn't exist, url: {url}, "
+            )
+
+    def upload_job_data(self, job_id, data: str, content_url=None):
+        """Upload job data"""
+        if not data:
+            raise SalesforceBulkV2LoadError("Data is required for ingest jobs")
+
+        # performance reduction here
+        data_size = len(data.encode("utf-8"))
+        if data_size > MAX_INGEST_JOB_FILE_SIZE:
+            raise SalesforceBulkV2LoadError(
+                f"Data size {data_size} exceeds the max file size accepted by "
+                "Bulk V2 (100 MB)"
+            )
+
+        url = (
+            content_url
+            or self._construct_request_url(job_id, False) + "/batches"
+        )
+        headers = self._get_headers(
+            self.CSV_CONTENT_TYPE, self.JSON_CONTENT_TYPE
+        )
+        result = call_salesforce(
+            url=url,
+            method="PUT",
+            session=self.session,
+            headers=headers,
+            data=data,
+        )
+        if result.status_code != http.CREATED:
+            raise SalesforceBulkV2LoadError(
+                f"Failed to upload job data. Error Code {result.status_code}. "
+                f"Response content: {result.content}"
+            )
+
+    def get_ingest_results(self, job_id, results_type):
+        """Get record results"""
+        url = self._construct_request_url(job_id, False) + "/" + results_type
+        headers = self._get_headers(
+            self.JSON_CONTENT_TYPE, self.CSV_CONTENT_TYPE
+        )
+        result = call_salesforce(
+            url=url, method="GET", session=self.session, headers=headers
+        )
+        return result.text
+
+    def download_ingest_results(
+        self, file, job_id, results_type, chunk_size=1024
+    ):
+        """Download record results to a file"""
+        url = self._construct_request_url(job_id, False) + "/" + results_type
+        headers = self._get_headers(
+            self.JSON_CONTENT_TYPE, self.CSV_CONTENT_TYPE
+        )
+        with closing(
+            call_salesforce(
+                url=url, method="GET", session=self.session, headers=headers
+            )
+        ) as result, open(file, "wb") as bos:
+            for chunk in result.iter_content(chunk_size=chunk_size):
+                bos.write(self.filter_null_bytes(chunk))
+
+        if not os.path.exists(file):
+            raise SalesforceBulkV2LoadError(
+                f"The IO/Error occured while verifying binary data. "
+                f"File {file} doesn't exist, url: {url}, "
+            )
+
+
+class SFBulk2Type:
+    """Interface to Bulk 2.0 API functions"""
+
+    def __init__(self, object_name, bulk2_url, headers, session):
+        """Initialize the instance with the given parameters.
+
+        Arguments:
+
+        * object_name -- the name of the type of SObject this represents,
+                         e.g. `Lead` or `Contact`
+        * bulk2_url -- API endpoint set in Salesforce instance
+        * headers -- bulk API headers
+        * session -- Custom requests session, created in calling code. This
+                     enables the use of requests Session features not otherwise
+                     exposed by simple_salesforce.
+        """
+        self.object_name = object_name
+        self.bulk2_url = bulk2_url
+        self.session = session
+        self.headers = headers
+        self._client = _Bulk2Client(object_name, bulk2_url, headers, session)
+
+    def _upload_data(
+        self,
+        operation,
+        data: Union[str, Tuple[int, str]],
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        external_id_field=None,
+        wait=5,
+    ) -> Dict:
+        """Upload data to Salesforce"""
+        if len(data) == 2:
+            total, data = data
+        else:
+            total = _count_csv(
+                data=data, line_ending=line_ending, skip_header=True
+            )
+        res = self._client.create_job(
+            operation,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            external_id_field=external_id_field,
+        )
+        job_id = res["id"]
+        try:
+            if res["state"] == JobState.open:
+                self._client.upload_job_data(job_id, data)
+                self._client.close_job(job_id)
+                self._client.wait_for_job(job_id, False, wait)
+                res = self._client.get_job(job_id, False)
+                return {
+                    "numberRecordsFailed": int(res["numberRecordsFailed"]),
+                    "numberRecordsProcessed": int(
+                        res["numberRecordsProcessed"]
+                    ),
+                    "numberRecordsTotal": int(total),
+                    "job_id": job_id,
+                }
+            raise SalesforceBulkV2LoadError(
+                f"Failed to upload job data. Response content: {res}"
+            )
+        except Exception:
+            res = self._client.get_job(job_id, False)
+            if res["state"] in (
+                JobState.upload_complete,
+                JobState.in_progress,
+                JobState.open,
+            ):
+                self._client.abort_job(job_id, False)
+            raise
+
+    def _upload_file(
+        self,
+        operation,
+        csv_file,
+        batch_size=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        external_id_field=None,
+        concurrency=1,
+        wait=5,
+    ) -> List[Dict]:
+        """Upload csv file to Salesforce"""
+        if not os.path.exists(csv_file):
+            raise SalesforceBulkV2LoadError(csv_file + " not found.")
+
+        if operation in (Operation.delete, Operation.hard_delete):
+            with open(csv_file, "r", encoding="utf-8") as bis:
+                header = (
+                    bis.readline()
+                    .rstrip()
+                    .split(_delimiter_char[column_delimiter])
+                )
+                if len(header) != 1:
+                    raise SalesforceBulkV2LoadError(
+                        f"InvalidBatch: The '{operation}' batch must contain "
+                        f"only ids, {header}"
+                    )
+
+        results = []
+        workers = min(concurrency, MAX_INGEST_JOB_PARALLELISM)
+        if workers == 1:
+            for data in _split_csv(csv_file, max_records=batch_size):
+                result = self._upload_data(
+                    operation,
+                    data,
+                    column_delimiter,
+                    line_ending,
+                    external_id_field,
+                    wait,
+                )
+                results.append(result)
+        else:
+            # OOM is possible if the file is too large
+            for chunks in chunked(
+                _split_csv(csv_file, max_records=batch_size), n=workers
+            ):
+                workers = min(workers, len(chunks))
+                with ThreadPoolExecutor(max_workers=workers) as pool:
+                    multi_thread_worker = partial(
+                        self._upload_data,
+                        operation,
+                        column_delimiter=column_delimiter,
+                        line_ending=line_ending,
+                        external_id_field=external_id_field,
+                        wait=wait,
+                    )
+                    _results = pool.map(multi_thread_worker, chunks)
+                results.extend(list(_results))
+        return results
+
+    def delete(
+        self,
+        csv_file,
+        batch_size=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        external_id_field=None,
+        wait=5,
+    ) -> List[Dict]:
+        """soft delete records"""
+        return self._upload_file(
+            Operation.delete,
+            csv_file,
+            batch_size=batch_size,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            external_id_field=external_id_field,
+            wait=wait,
+        )
+
+    def insert(
+        self,
+        csv_file,
+        batch_size=None,
+        concurrency=1,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> List[Dict]:
+        """insert records"""
+        return self._upload_file(
+            Operation.insert,
+            csv_file,
+            batch_size=batch_size,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            concurrency=concurrency,
+            wait=wait,
+        )
+
+    def upsert(
+        self,
+        csv_file,
+        external_id_field,
+        batch_size=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> List[Dict]:
+        """upsert records based on a unique identifier"""
+        return self._upload_file(
+            Operation.upsert,
+            csv_file,
+            batch_size=batch_size,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            external_id_field=external_id_field,
+            wait=wait,
+        )
+
+    def update(
+        self,
+        csv_file,
+        batch_size=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> List[Dict]:
+        """update records"""
+        return self._upload_file(
+            Operation.update,
+            csv_file,
+            batch_size=batch_size,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            wait=wait,
+        )
+
+    def hard_delete(
+        self,
+        csv_file,
+        batch_size=None,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> List[Dict]:
+        """hard delete records"""
+        return self._upload_file(
+            Operation.hard_delete,
+            csv_file,
+            batch_size=batch_size,
+            column_delimiter=column_delimiter,
+            line_ending=line_ending,
+            wait=wait,
+        )
+
+    def query(
+        self,
+        query,
+        max_records=DEFAULT_QUERY_PAGE_SIZE,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> Generator[str, None, None]:
+        """bulk 2.0 query
+
+        Arguments:
+        * query -- SOQL query
+        * max_records -- max records to retrieve per batch, default 50000
+
+        Returns:
+        * locator  -- the locator for the next set of results
+        * number_of_records -- the number of records in this set
+        * records -- records in this set
+        """
+        res = self._client.create_job(
+            Operation.query, query, column_delimiter, line_ending
+        )
+        job_id = res["id"]
+        self._client.wait_for_job(job_id, True, wait)
+
+        locator = "INIT"
+        while locator:
+            if locator == "INIT":
+                locator = ""
+            result = self._client.get_query_results(
+                job_id, locator, max_records
+            )
+            locator = result["locator"]
+            yield result["records"]
+
+    def download(
+        self,
+        query,
+        path,
+        max_records=DEFAULT_QUERY_PAGE_SIZE,
+        column_delimiter=ColumnDelimiter.COMMA,
+        line_ending=LineEnding.LF,
+        wait=5,
+    ) -> List[Dict]:
+        """bulk 2.0 query stream to file, avoiding high memory usage
+
+        Arguments:
+        * query -- SOQL query
+        * max_records -- max records to retrieve per batch, default 50000
+
+        Returns:
+        * locator  -- the locator for the next set of results
+        * number_of_records -- the number of records in this set
+        * file -- downloaded file
+        """
+        if not os.path.exists(path):
+            raise SalesforceBulkV2LoadError(f"Path does not exist: {path}")
+
+        res = self._client.create_job(
+            Operation.query, query, column_delimiter, line_ending
+        )
+        job_id = res["id"]
+        self._client.wait_for_job(job_id, True, wait)
+
+        results = []
+        locator = "INIT"
+        while locator:
+            if locator == "INIT":
+                locator = ""
+            result = self._client.download_job_data(
+                path, job_id, locator, max_records
+            )
+            locator = result["locator"]
+            results.append(result)
+        return results
+
+    def _retrieve_ingest_records(self, job_id, results_type, file=None):
+        """Retrieve the results of an ingest job"""
+        if not file:
+            return self._client.get_ingest_results(job_id, results_type)
+        self._client.download_ingest_results(file, job_id, results_type)
+        return ""
+
+    def get_failed_records(self, job_id, file=None):
+        """Get failed record results
+
+        Results Property:
+            sf__Id:	[string] ID of the record
+            sf__Error:	[Error]	Error code and message
+            Fields from the original CSV request data:	various
+        """
+        return self._retrieve_ingest_records(job_id, ResultsType.failed, file)
+
+    def get_unprocessed_records(self, job_id, file=None):
+        """Get unprocessed record results
+
+        Results Property:
+            Fields from the original CSV request data:	[various]
+        """
+        return self._retrieve_ingest_records(
+            job_id, ResultsType.unprocessed, file
+        )
+
+    def get_successful_records(self, job_id, file=None):
+        """Get successful record results.
+
+        Results Property:
+            sf__Id:	[string] ID of the record
+            sf__Created: [boolean] Indicates if the record was created
+            Fields from the original CSV request data:	[various]
+        """
+        return self._retrieve_ingest_records(
+            job_id, ResultsType.successful, file
+        )
diff --git a/simple_salesforce/exceptions.py b/simple_salesforce/exceptions.py
index ac90f652..be70bd07 100644
--- a/simple_salesforce/exceptions.py
+++ b/simple_salesforce/exceptions.py
@@ -85,6 +85,7 @@ def __str__(self):
         return self.message.format(name=self.resource_name,
                                    content=self.content)
 
+
 class SalesforceAuthenticationFailed(SalesforceError):
     """
     Thrown to indicate that authentication with Salesforce failed.
@@ -98,7 +99,7 @@ def __init__(self, code, message):
         self.message = message
 
     def __str__(self):
-        return '{code}: {message}'.format(code=self.code, message=self.message)
+        return f'{self.code}: {self.message}'
 
 
 class SalesforceGeneralError(SalesforceError):
@@ -110,3 +111,19 @@ class SalesforceGeneralError(SalesforceError):
 
     def __str__(self):
         return self.message.format(status=self.status, content=self.content)
+
+
+class SalesforceOperationError(Exception):
+    """Base error for Bulk API 2.0 operations"""
+
+
+class SalesforceBulkV2LoadError(SalesforceOperationError):
+    """
+    Error occurred during bulk 2.0 load
+    """
+
+
+class SalesforceBulkV2ExtractError(SalesforceOperationError):
+    """
+    Error occurred during bulk 2.0 extract
+    """
diff --git a/simple_salesforce/login.py b/simple_salesforce/login.py
index ed6ad796..74ccaedc 100644
--- a/simple_salesforce/login.py
+++ b/simple_salesforce/login.py
@@ -7,18 +7,19 @@
 
 import warnings
 from datetime import datetime, timedelta, timezone
-from html import escape
+from html import escape, unescape
 from json.decoder import JSONDecodeError
+from pathlib import Path
 
 import requests
-from authlib.jose import jwt
+import jwt
 
 from .api import DEFAULT_API_VERSION
 from .exceptions import SalesforceAuthenticationFailed
 from .util import getUniqueElementValueFromXmlString
 
 
-# pylint: disable=invalid-name,too-many-arguments,too-many-locals
+# pylint: disable=invalid-name,too-many-arguments,too-many-locals,too-many-branches
 def SalesforceLogin(
         username=None,
         password=None,
@@ -67,18 +68,18 @@ def SalesforceLogin(
     if domain is None:
         domain = 'login'
 
-    soap_url = 'https://{domain}.salesforce.com/services/Soap/u/{sf_version}'
+    if sf_version.startswith("v"):
+        error_msg = (
+            f"Invalid sf_version specified ({sf_version}). Version should not "
+            "contain a leading 'v'"
+        )
+        raise ValueError(error_msg)
 
     if client_id:
-        client_id = "{prefix}/{app_name}".format(
-            prefix=DEFAULT_CLIENT_ID_PREFIX,
-            app_name=client_id)
+        client_id = f'{DEFAULT_CLIENT_ID_PREFIX}/{client_id}'
     else:
         client_id = DEFAULT_CLIENT_ID_PREFIX
 
-    soap_url = soap_url.format(domain=domain,
-                               sf_version=sf_version)
-
     # pylint: disable=E0012,deprecated-method
     username = escape(username) if username else None
     password = escape(password) if password else None
@@ -86,32 +87,45 @@ def SalesforceLogin(
     # Check if token authentication is used
     if security_token is not None:
         # Security Token Soap request body
-        login_soap_request_body = """<?xml version="1.0" encoding="utf-8" ?>
-        <env:Envelope
-                xmlns:xsd="http://www.w3.org/2001/XMLSchema"
-                xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-                xmlns:env="http://schemas.xmlsoap.org/soap/envelope/"
-                xmlns:urn="urn:partner.soap.sforce.com">
-            <env:Header>
-                <urn:CallOptions>
-                    <urn:client>{client_id}</urn:client>
-                    <urn:defaultNamespace>sf</urn:defaultNamespace>
-                </urn:CallOptions>
-            </env:Header>
-            <env:Body>
-                <n1:login xmlns:n1="urn:partner.soap.sforce.com">
-                    <n1:username>{username}</n1:username>
-                    <n1:password>{password}{token}</n1:password>
-                </n1:login>
-            </env:Body>
-        </env:Envelope>""".format(
-            username=username, password=password, token=security_token,
-            client_id=client_id)
+        login_soap_request_body = f"""<?xml version="1.0" encoding="utf-8" ?>
+<env:Envelope
+        xmlns:xsd="http://www.w3.org/2001/XMLSchema"
+        xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+        xmlns:env="http://schemas.xmlsoap.org/soap/envelope/"
+        xmlns:urn="urn:partner.soap.sforce.com">
+    <env:Header>
+        <urn:CallOptions>
+            <urn:client>{client_id}</urn:client>
+            <urn:defaultNamespace>sf</urn:defaultNamespace>
+        </urn:CallOptions>
+    </env:Header>
+    <env:Body>
+        <n1:login xmlns:n1="urn:partner.soap.sforce.com">
+            <n1:username>{username}</n1:username>
+            <n1:password>{password}{security_token}</n1:password>
+        </n1:login>
+    </env:Body>
+</env:Envelope>"""
+
+    elif username is not None and \
+            password is not None and \
+            consumer_key is not None and \
+            consumer_secret is not None:
+        token_data = {
+            'grant_type': 'password',
+            'client_id': consumer_key,
+            'client_secret': consumer_secret,
+            'username': unescape(username),
+            'password': unescape(password) if password else None
+            }
+        return token_login(
+            f'https://{domain}.salesforce.com/services/oauth2/token',
+            token_data, domain, consumer_key,
+            None, proxies, session)
 
     elif username is not None and \
             consumer_key is not None and \
             consumer_secret is not None:
-        header = {'Content-Type': 'application/json'}
         payload = {
             'grant_type': 'password',
             'client_id': consumer_key,
@@ -120,91 +134,77 @@ def SalesforceLogin(
             'password': password
             }
 
-        login_token_request_data = {
-            'grant_type': 'password'
-            }
-
         return token_login(
-            'https://{domain}.salesforce.com/services/oauth2/token'.format(
-                domain=domain),
+            f'https://{domain}.salesforce.com/services/oauth2/token',
             payload, domain, consumer_key,
             None, proxies, session)
 
     # Check if IP Filtering is used in conjunction with organizationId
     elif organizationId is not None:
         # IP Filtering Login Soap request body
-        login_soap_request_body = """<?xml version="1.0" encoding="utf-8" ?>
-        <soapenv:Envelope
-                xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
-                xmlns:urn="urn:partner.soap.sforce.com">
-            <soapenv:Header>
-                <urn:CallOptions>
-                    <urn:client>{client_id}</urn:client>
-                    <urn:defaultNamespace>sf</urn:defaultNamespace>
-                </urn:CallOptions>
-                <urn:LoginScopeHeader>
-                    <urn:organizationId>{organizationId}</urn:organizationId>
-                </urn:LoginScopeHeader>
-            </soapenv:Header>
-            <soapenv:Body>
-                <urn:login>
-                    <urn:username>{username}</urn:username>
-                    <urn:password>{password}</urn:password>
-                </urn:login>
-            </soapenv:Body>
-        </soapenv:Envelope>""".format(
-            username=username, password=password, organizationId=organizationId,
-            client_id=client_id)
+        login_soap_request_body = f"""<?xml version="1.0" encoding="utf-8" ?>
+<soapenv:Envelope
+        xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
+        xmlns:urn="urn:partner.soap.sforce.com">
+    <soapenv:Header>
+        <urn:CallOptions>
+            <urn:client>{client_id}</urn:client>
+            <urn:defaultNamespace>sf</urn:defaultNamespace>
+        </urn:CallOptions>
+        <urn:LoginScopeHeader>
+            <urn:organizationId>{organizationId}</urn:organizationId>
+        </urn:LoginScopeHeader>
+    </soapenv:Header>
+    <soapenv:Body>
+        <urn:login>
+            <urn:username>{username}</urn:username>
+            <urn:password>{password}</urn:password>
+        </urn:login>
+    </soapenv:Body>
+</soapenv:Envelope>"""
     elif username is not None and password is not None:
         # IP Filtering for non self-service users
-        login_soap_request_body = """<?xml version="1.0" encoding="utf-8" ?>
-        <soapenv:Envelope
-                xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
-                xmlns:urn="urn:partner.soap.sforce.com">
-            <soapenv:Header>
-                <urn:CallOptions>
-                    <urn:client>{client_id}</urn:client>
-                    <urn:defaultNamespace>sf</urn:defaultNamespace>
-                </urn:CallOptions>
-            </soapenv:Header>
-            <soapenv:Body>
-                <urn:login>
-                    <urn:username>{username}</urn:username>
-                    <urn:password>{password}</urn:password>
-                </urn:login>
-            </soapenv:Body>
-        </soapenv:Envelope>""".format(
-            username=username, password=password, client_id=client_id)
+        login_soap_request_body = f"""<?xml version="1.0" encoding="utf-8" ?>
+<soapenv:Envelope
+        xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/"
+        xmlns:urn="urn:partner.soap.sforce.com">
+    <soapenv:Header>
+        <urn:CallOptions>
+            <urn:client>{client_id}</urn:client>
+            <urn:defaultNamespace>sf</urn:defaultNamespace>
+        </urn:CallOptions>
+    </soapenv:Header>
+    <soapenv:Body>
+        <urn:login>
+            <urn:username>{username}</urn:username>
+            <urn:password>{password}</urn:password>
+        </urn:login>
+    </soapenv:Body>
+</soapenv:Envelope>"""
     elif username is not None and \
             consumer_key is not None and \
             (privatekey_file is not None or privatekey is not None):
-        header = {'alg': 'RS256'}
         expiration = datetime.now(timezone.utc) + timedelta(minutes=3)
         payload = {
             'iss': consumer_key,
             'sub': username,
-            'aud': 'https://{domain}.salesforce.com'.format(domain=domain),
-            'exp': '{exp:.0f}'.format(
-                exp=expiration.timestamp()
-                )
+            'aud': f'https://{domain}.salesforce.com',
+            'exp': f'{expiration.timestamp():.0f}'
             }
         if privatekey_file is not None:
-            with open(privatekey_file, 'rb') as key_file:
-                key = key_file.read()
+            key = Path(privatekey_file).read_bytes()
         else:
             key = privatekey
+        assertion = jwt.encode(payload, key, algorithm='RS256')
 
-        assertion = jwt.encode(header, payload, key)
-
-        login_token_request_data = {
+        token_data = {
             'grant_type': 'urn:ietf:params:oauth:grant-type:jwt-bearer',
             'assertion': assertion
             }
 
         return token_login(
-            'https://{domain}.salesforce.com/services/oauth2/token'.format(
-                domain=domain),
-            login_token_request_data, domain, consumer_key,
+            f'https://{domain}.salesforce.com/services/oauth2/token',
+            token_data, domain, consumer_key,
             None, proxies, session)
     else:
         except_code = 'INVALID AUTH'
@@ -214,6 +214,7 @@ def SalesforceLogin(
         )
         raise SalesforceAuthenticationFailed(except_code, except_msg)
 
+    soap_url = f'https://{domain}.salesforce.com/services/Soap/u/{sf_version}'
     login_soap_request_headers = {
         'content-type': 'text/xml',
         'charset': 'UTF-8',
@@ -267,19 +268,16 @@ def token_login(token_url, token_data, domain, consumer_key,
         except_code = json_response.get('error')
         except_msg = json_response.get('error_description')
         if except_msg == "user hasn't approved this consumer":
-            auth_url = 'https://{domain}.salesforce.com/services/oauth2/' \
+            auth_url = f'https://{domain}.salesforce.com/services/oauth2/' \
                        'authorize?response_type=code&client_id=' \
-                       '{consumer_key}&redirect_uri=<approved URI>'.format(
-                domain=domain,
-                consumer_key=consumer_key
-                )
-            warnings.warn("""
+                       f'{consumer_key}&redirect_uri=<approved URI>'
+            warnings.warn(f"""
     If your connected app policy is set to "All users may
     self-authorize", you may need to authorize this
     application first. Browse to
-    %s
+    {auth_url}
     in order to Allow Access. Check first to ensure you have a valid
-    <approved URI>.""" % auth_url)
+    <approved URI>.""")
         raise SalesforceAuthenticationFailed(except_code, except_msg)
 
     access_token = json_response.get('access_token')
diff --git a/simple_salesforce/metadata.py b/simple_salesforce/metadata.py
index 3f2fe3a8..157808ee 100644
--- a/simple_salesforce/metadata.py
+++ b/simple_salesforce/metadata.py
@@ -1,6 +1,6 @@
 """ Class to work with Salesforce Metadata API """
-import os
 from base64 import b64encode, b64decode
+from pathlib import Path
 from xml.etree import ElementTree as ET
 from .util import call_salesforce
 from .messages import DEPLOY_MSG, CHECK_DEPLOY_STATUS_MSG, \
@@ -44,10 +44,9 @@ def _handle_api_response(response):
         err_string = ""
         for result in response:
             if not result.success:
-                err_string += "\n{}: ".format(result.fullName)
+                err_string += f'\n{result.fullName}: '
                 for error in result.errors:
-                    err_string += "({}, {}), ".format(error.statusCode,
-                                                      error.message)
+                    err_string += f'({error.statusCode}, {error.message}), '
         if err_string:
             raise Exception(err_string)
 
@@ -132,7 +131,7 @@ def upsert(self, metadata):
                          mix of both types.
         :type metadata: list
         """
-        response = self._service.updateMetadata(metadata, _soapheaders=[
+        response = self._service.upsertMetadata(metadata, _soapheaders=[
             self._session_header])
         self._handle_api_response(response)
 
@@ -176,7 +175,7 @@ def describe(self):
         :returns: DescribeValueTypeResult
         """
         return self._service.describeValueType(
-            "{{http://soap.sforce.com/2006/04/metadata}}{}".format(self._name),
+            f'{{http://soap.sforce.com/2006/04/metadata}}{self._name}',
             _soapheaders=[self._session_header])
 
 
@@ -200,9 +199,8 @@ def __init__(self, session, session_id, instance, metadata_url, headers,
         self.headers = headers
         self._api_version = api_version
         self._deploy_zip = None
-        wsdl_path = os.path.join(os.path.dirname(os.path.abspath(__file__)),
-                                 'metadata.wsdl')
-        self._client = Client(os.path.join('simple_salesforce', wsdl_path),
+        wsdl_path = Path(__file__).parent / 'metadata.wsdl'
+        self._client = Client(wsdl_path.absolute().as_uri(),
                               settings=Settings(strict=False,
                                                 xsd_ignore_sequence_order=True))
         self._service = self._client.create_service(
@@ -283,14 +281,14 @@ def deploy(self, zipfile, sandbox, **kwargs):
             attributes['rollbackOnError'] = True
 
         if testLevel:
-            test_level = "<met:testLevel>%s</met:testLevel>" % testLevel
+            test_level = f'<met:testLevel>{testLevel}</met:testLevel>'
             attributes['testLevel'] = test_level
 
         tests_tag = ''
         if tests and \
                 str(testLevel).lower() == 'runspecifiedtests':
             for test in tests:
-                tests_tag += '<met:runTests>%s</met:runTests>\n' % test
+                tests_tag += f'<met:runTests>{test}</met:runTests>\n'
             attributes['tests'] = tests_tag
 
         request = DEPLOY_MSG.format(**attributes)
@@ -322,16 +320,11 @@ def _read_deploy_zip(zipfile):
         :rtype:
         """
         if hasattr(zipfile, 'read'):
-            file = zipfile
-            file.seek(0)
-            should_close = False
+            zipfile.seek(0)
+            raw = zipfile.read()
         else:
-            file = open(zipfile, 'rb')
-            should_close = True
-        raw = file.read()
-        if should_close:
-            file.close()
-        return b64encode(raw).decode("utf-8")
+            raw = Path(zipfile).read_bytes()
+        return b64encode(raw).decode()
 
     def _retrieve_deploy_result(self, async_process_id, **kwargs):
         """ Retrieves status for specified deployment id
@@ -368,7 +361,7 @@ def _retrieve_deploy_result(self, async_process_id, **kwargs):
             'soapenv:Body/mt:checkDeployStatusResponse/mt:result',
             self._XML_NAMESPACES)
         if result is None:
-            raise Exception("Result node could not be found: %s" % res.text)
+            raise Exception(f"Result node could not be found: {res.text}")
 
         return result
 
@@ -456,8 +449,8 @@ def download_unit_test_logs(self, async_process_id):
         """ Downloads Apex logs for unit tests executed during specified
         deployment """
         result = self._retrieve_deploy_result(async_process_id)
-        print("response: %s" % ET.tostring(result, encoding="us-ascii",
-                                           method="xml"))
+        print("response:", ET.tostring(result, encoding="us-ascii",
+                                       method="xml"))
 
     def retrieve(self, async_process_id, **kwargs):
         """ Submits retrieve request """
@@ -475,10 +468,8 @@ def retrieve(self, async_process_id, **kwargs):
                     members = kwargs.get('unpackaged')[metadata_type]
                     unpackaged += '<types>'
                     for member in members:
-                        unpackaged += '<members>{member}</members>'.format(
-                            member=member)
-                    unpackaged += '<name>{metadata_type}</name></types>'.format(
-                        metadata_type=metadata_type)
+                        unpackaged += f'<members>{member}</members>'
+                    unpackaged += f'<name>{metadata_type}</name></types>'
                 else:
                     raise TypeError('unpackaged metadata types must be a dict')
 
@@ -538,7 +529,7 @@ def retrieve_retrieve_result(self, async_process_id, include_zip, **kwargs):
             'soapenv:Body/mt:checkRetrieveStatusResponse/mt:result',
             self._XML_NAMESPACES)
         if result is None:
-            raise Exception("Result node could not be found: %s" % res.text)
+            raise Exception(f"Result node could not be found: {res.text}")
 
         return result
 
diff --git a/simple_salesforce/util.py b/simple_salesforce/util.py
index 382006aa..55b0d82c 100644
--- a/simple_salesforce/util.py
+++ b/simple_salesforce/util.py
@@ -34,15 +34,9 @@ def getUniqueElementValueFromXmlString(xmlString, elementName):
 def date_to_iso8601(date):
     """Returns an ISO8601 string from a date"""
     datetimestr = date.strftime('%Y-%m-%dT%H:%M:%S')
-    timezone_sign = date.strftime('%z')[0:1]
-    timezone_str = '%s:%s' % (
-        date.strftime('%z')[1:3],
-        date.strftime('%z')[3:5],
-    )
+    timezonestr = date.strftime('%z')
     return (
-        '{datetimestr}{tzsign}{timezone}'.format(
-            datetimestr=datetimestr, tzsign=timezone_sign, timezone=timezone_str
-        )
+        f'{datetimestr}{timezonestr[0:3]}:{timezonestr[3:5]}'
         .replace(':', '%3A')
         .replace('+', '%2B')
     )


Test patch:
diff --git a/simple_salesforce/tests/__init__.py b/simple_salesforce/tests/__init__.py
index 49727cea..2a57ac91 100644
--- a/simple_salesforce/tests/__init__.py
+++ b/simple_salesforce/tests/__init__.py
@@ -11,16 +11,16 @@
     "https": "http://10.10.1.10:1080",
 }
 
-LOGIN_RESPONSE_SUCCESS = """<?xml version="1.0" encoding="UTF-8"?>
+LOGIN_RESPONSE_SUCCESS = f"""<?xml version="1.0" encoding="UTF-8"?>
 <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/" xmlns="urn:enterprise.soap.sforce.com" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
    <soapenv:Body>
       <loginResponse>
          <result>
-            <metadataServerUrl>%s</metadataServerUrl>
+            <metadataServerUrl>{METADATA_URL}</metadataServerUrl>
             <passwordExpired>false</passwordExpired>
             <sandbox>false</sandbox>
-            <serverUrl>%s</serverUrl>
-            <sessionId>%s</sessionId>
+            <serverUrl>{SERVER_URL}</serverUrl>
+            <sessionId>{SESSION_ID}</sessionId>
             <userId>005i0000002MUqLAAW</userId>
             <userInfo>
                <accessibilityMode>false</accessibilityMode>
@@ -50,15 +50,15 @@
       </loginResponse>
    </soapenv:Body>
 </soapenv:Envelope>
-""" % (METADATA_URL, SERVER_URL, SESSION_ID)
+"""
 
-TOKEN_LOGIN_RESPONSE_SUCCESS = """{
-    "access_token": "%s",
+TOKEN_LOGIN_RESPONSE_SUCCESS = f"""{{
+    "access_token": "{SESSION_ID}",
     "scope": "web api",
-    "instance_url": "%s",
-    "id": "%s",
+    "instance_url": "{INSTANCE_URL}",
+    "id": "{TOKEN_ID}",
     "token_type": "Bearer"
-}""" % (SESSION_ID, INSTANCE_URL, TOKEN_ID)
+}}"""
 
 TOKEN_WARNING = """
     If your connected app policy is set to "All users may
@@ -96,6 +96,12 @@
 
 BULK_HEADERS = {
     'Content-Type': 'application/json',
-    'X-SFDC-Session': "%s" % SESSION_ID,
+    'X-SFDC-Session': SESSION_ID,
+    'X-PrettyPrint': '1'
+}
+
+BULKv2_HEADERS = {
+    'Content-Type': 'application/json',
+    "Authorization": 'Bearer ' + SESSION_ID,
     'X-PrettyPrint': '1'
 }
diff --git a/simple_salesforce/tests/sample.zip b/simple_salesforce/tests/sample.zip
new file mode 100644
index 00000000..15cb0ecb
Binary files /dev/null and b/simple_salesforce/tests/sample.zip differ
diff --git a/simple_salesforce/tests/test_api.py b/simple_salesforce/tests/test_api.py
index 2bf11c9e..53c6ab4c 100644
--- a/simple_salesforce/tests/test_api.py
+++ b/simple_salesforce/tests/test_api.py
@@ -6,6 +6,7 @@
 import decimal
 from collections import OrderedDict
 from datetime import datetime
+from pathlib import Path
 from unittest.mock import patch
 
 import requests
@@ -534,7 +535,7 @@ def test_custom_version_success(self):
             version=expected_version)
 
         self.assertEqual(
-            client.base_url.split('/')[-2], 'v%s' % expected_version)
+            client.base_url.split('/')[-2], f'v{expected_version}')
 
     def test_shared_session_to_sftype(self):
         """Test Salesforce and SFType instances share default `Session`"""
@@ -811,8 +812,7 @@ def test_api_limits(self):
         self.assertEqual(result, tests.ORGANIZATION_LIMITS_RESPONSE)
 
     @responses.activate
-    @patch("simple_salesforce.metadata.SfdcMetadataApi._read_deploy_zip")
-    def test_md_deploy_success(self, mock_read_zip):
+    def test_md_deploy_success(self):
         """"
         Test method for metadata deployment
         """
@@ -837,13 +837,44 @@ def test_md_deploy_success(self, mock_read_zip):
         client = Salesforce(session_id=tests.SESSION_ID,
                             instance=tests.INSTANCE_URL,
                             session=session)
-        result = client.deploy("path/to/fake/zip.zip", sandbox=False)
+        zip_path = Path(__file__).parent / 'sample.zip'
+        result = client.deploy(str(zip_path), sandbox=False)
         self.assertEqual(result.get('asyncId'), "0Af3B00001CMyfASAT")
         self.assertEqual(result.get('state'), "Queued")
 
     @responses.activate
-    @patch("simple_salesforce.metadata.SfdcMetadataApi._read_deploy_zip")
-    def test_md_deploy_failed_status_code(self, mock_read_zip):
+    def test_md_deploy_success_open(self):
+        """"
+        Test method for metadata deployment
+        """
+        #pylint: disable=unused-argument
+        mock_response = '<?xml version="1.0" ' \
+                        'encoding="UTF-8"?><soapenv:Envelope ' \
+                        'xmlns:soapenv="http://schemas.xmlsoap.org/soap' \
+                        '/envelope/" ' \
+                        'xmlns="http://soap.sforce.com/2006/04/metadata' \
+                        '"><soapenv:Body><deployResponse><result><done' \
+                        '>false</done><id>0Af3B00001CMyfASAT</id><state' \
+                        '>Queued</state></result></deployResponse></soapenv' \
+                        ':Body></soapenv:Envelope>'
+        responses.add(
+            responses.POST,
+            re.compile(r'^https://.*/deployRequest'),
+            body=mock_response,
+            status=200
+            )
+
+        session = requests.Session()
+        client = Salesforce(session_id=tests.SESSION_ID,
+                            instance_url=tests.INSTANCE_URL,
+                            session=session)
+        zip_path = Path(__file__).parent / 'sample.zip'
+        with zip_path.open('rb') as zipfile:
+            client.deploy(zipfile, sandbox=False)
+            self.assertFalse(zipfile.closed)
+
+    @responses.activate
+    def test_md_deploy_failed_status_code(self):
         """"
         Test method for metadata deployment
         """
@@ -859,8 +890,9 @@ def test_md_deploy_failed_status_code(self, mock_read_zip):
         client = Salesforce(session_id=tests.SESSION_ID,
                             instance=tests.INSTANCE_URL,
                             session=session)
+        zip_path = Path(__file__).parent / 'sample.zip'
         with self.assertRaises(Exception):
-            client.deploy("path/to/fake/zip.zip", sandbox=False)
+            client.deploy(str(zip_path), sandbox=False)
 
     @responses.activate
     def test_check_status_pending(self):
@@ -1182,7 +1214,7 @@ def test_oauth2_without_json_result(self):
         """Test oauth2 endpoint returns no result"""
         responses.add(
             responses.POST,
-            re.compile(r'^https://.*/services/oauth2/revoke\?token=.+$$'),
+            re.compile(r'^https://.*/services/oauth2/revoke\?token=.+$'),
             body='{}',
             status=http.OK,
             content_type='')
diff --git a/simple_salesforce/tests/test_bulk.py b/simple_salesforce/tests/test_bulk.py
index 2431329b..4460cdf0 100644
--- a/simple_salesforce/tests/test_bulk.py
+++ b/simple_salesforce/tests/test_bulk.py
@@ -1,13 +1,19 @@
 """Test for bulk.py"""
 import http.client as http
+import json
+
+import itertools
+import random
 import re
 import unittest
+from unittest import mock
 from unittest.mock import patch
 
 import requests
 import responses
 from simple_salesforce import tests
 from simple_salesforce.api import Salesforce
+from simple_salesforce.bulk import SFBulkType
 from simple_salesforce.exceptions import SalesforceGeneralError
 
 
@@ -83,7 +89,7 @@ def test_delete(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -96,7 +102,7 @@ def test_delete(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -141,7 +147,7 @@ def test_insert(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -154,7 +160,7 @@ def test_insert(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -206,7 +212,7 @@ def test_upsert(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -219,7 +225,7 @@ def test_upsert(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -273,7 +279,7 @@ def test_update(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -286,7 +292,7 @@ def test_update(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -340,7 +346,7 @@ def test_query(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -353,7 +359,7 @@ def test_query(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -418,7 +424,7 @@ def test_query_fail(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -431,7 +437,7 @@ def test_query_fail(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -465,7 +471,7 @@ def test_query_lazy(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -478,7 +484,7 @@ def test_query_lazy(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -546,7 +552,7 @@ def test_query_all(self):
             re.compile(r'^https://[^/job].*/job$'),
             body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
             '"contentType": "JSON","id": "Job-1","object": "Contact",'
-            '"operation": "%s","state": "Open"}' % operation,
+            f'"operation": "{operation}","state": "Open"}}',
             status=http.OK)
         responses.add(
             responses.POST,
@@ -559,7 +565,7 @@ def test_query_all(self):
             re.compile(r'^https://[^/job].*/job/Job-1$'),
             body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
             '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
-            '"operation" : "%s","state" : "Closed"}' % operation,
+            f'"operation" : "{operation}","state" : "Closed"}}',
             status=http.OK
         )
         responses.add(
@@ -612,3 +618,81 @@ def test_query_all(self):
                             session=session)
         contact = client.bulk.Contact.query_all(data)
         self.assertEqual(self.expected_query, contact)
+
+    @responses.activate
+    @mock.patch('simple_salesforce.bulk.SFBulkType._add_autosized_batches')
+    def test_bulk_operation_auto_batch_size(self, add_autosized_batches):
+        """Test that batch_size="auto" leads to using _add_autosized_batches"""
+        session = requests.Session()
+        client = Salesforce(session_id=tests.SESSION_ID,
+                            instance_url=tests.SERVER_URL,
+                            session=session)
+
+        operation = "update"
+
+        responses.add(
+            responses.POST,
+            re.compile(r'^https://[^/job].*/job$'),
+            body='{"apiVersion": 42.0, "concurrencyMode": "Parallel",'
+            '"contentType": "JSON","id": "Job-1","object": "Contact",'
+            f'"operation": "{operation}","state": "Open"}}',
+            status=http.OK)
+        responses.add(
+            responses.POST,
+            re.compile(r'^https://[^/job].*/job/Job-1$'),
+            body='{"apiVersion" : 42.0, "concurrencyMode" : "Parallel",'
+            '"contentType" : "JSON","id" : "Job-1","object" : "Contact",'
+            f'"operation" : "{operation}","state" : "Closed"}}',
+            status=http.OK
+        )
+        data = [{
+            'AccountId': 'ID-1',
+            'Email': 'contact1@example.com',
+            'FirstName': 'Bob',
+            'LastName': 'x'
+        }]
+        add_autosized_batches.return_value = []
+        client.bulk.Contact._bulk_operation(  # pylint: disable=protected-access
+            operation, data, batch_size="auto"
+        )
+        add_autosized_batches.assert_called_once_with(
+            job='Job-1', data=data, operation=operation
+        )
+
+    @mock.patch('simple_salesforce.bulk.SFBulkType._add_batch')
+    def test_add_autosized_batches(self, add_batch):
+        """Test that _add_autosized_batches batches all records correctly"""
+        # _add_autosized_batches passes the return values from add_batch, so we
+        # can pass the data it was given back so that we can test it
+        add_batch.side_effect = lambda job_id, data, operation: data
+        sf_bulk_type = SFBulkType(None, None, None, None)
+        data = [
+            # Expected serialized record size of 13 to 1513. Idea is that
+            # earlier record batches are split on record count, whereas later
+            # batches are split for hitting the byte limit.
+            {'key': 'value' * random.randint(0, i // 50)}
+            for i in range(30000)
+        ]
+        result = sf_bulk_type._add_autosized_batches(  # pylint: disable=protected-access
+            data=data, operation="update", job="Job-1"
+        )
+        reconstructed_data = list(itertools.chain(*result))
+        # all data was put in a batch
+        self.assertEqual(len(data), len(reconstructed_data))
+        self.assertEqual(data, list(itertools.chain(*result)))
+
+        for i, batch in enumerate(result):
+            record_count = len(batch)
+            size_in_bytes = len(json.dumps(batch))
+            is_last_batch = i == len(result) - 1
+            # Check that all batches are within limits
+            self.assertLessEqual(record_count, 10_000)
+            self.assertLessEqual(size_in_bytes, 10_000_000)
+            # ... and that - except for the last batch - all batches have maxed
+            # out one of the two limits
+            self.assertTrue(
+                is_last_batch or
+                record_count == 10_000 or
+                (size_in_bytes + len(json.dumps(result[i + 1][0])) + 2
+                    > 10_000_000)
+            )
diff --git a/simple_salesforce/tests/test_bulk2.py b/simple_salesforce/tests/test_bulk2.py
new file mode 100644
index 00000000..7953cff1
--- /dev/null
+++ b/simple_salesforce/tests/test_bulk2.py
@@ -0,0 +1,571 @@
+"""Test for bulk2.py"""
+import csv
+import http.client as http
+import json
+import os
+import re
+import tempfile
+import textwrap
+import unittest
+from contextlib import contextmanager
+from functools import partial
+from unittest.mock import patch
+
+import requests
+import responses
+
+from simple_salesforce import tests
+from simple_salesforce.api import Salesforce
+from simple_salesforce.bulk2 import JobState, Operation
+
+# pylint: disable=line-too-long,missing-docstring
+
+to_body = partial(json.dumps, ensure_ascii=False)
+
+
+@contextmanager
+def to_csv_file(data):
+    """Create a temporary csv file from a list of dicts, delete it when done."""
+    temp_file = None
+    try:
+        temp_file = tempfile.NamedTemporaryFile(mode="w", delete=False)
+        writer = csv.DictWriter(temp_file, fieldnames=data[0].keys())
+        writer.writeheader()
+        writer.writerows(data)
+        temp_file.flush()
+        yield temp_file.name
+    finally:
+        if temp_file is not None:
+            os.remove(temp_file.name)
+
+
+@contextmanager
+def temp_csv_file():
+    temp_file = None
+    try:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        yield temp_file.name
+    finally:
+        if temp_file is not None:
+            os.remove(temp_file.name)
+
+
+class TestSFBulk2Handler(unittest.TestCase):
+    """Test for SFBulkHandler"""
+
+    def test_bulk2_handler(self):
+        """Test bulk2 handler creation"""
+        session = requests.Session()
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=session,
+        )
+        bulk2_handler = client.bulk2
+
+        self.assertIs(tests.SESSION_ID, bulk2_handler.session_id)
+        self.assertIs(client.bulk2_url, bulk2_handler.bulk2_url)
+        self.assertEqual(tests.BULKv2_HEADERS, bulk2_handler.headers)
+
+
+def ingest_responses(operation, processed, failed=0):
+    """Mock responses for bulk2 ingest jobs"""
+    responses.add(
+        responses.POST,
+        re.compile(r"^https://.*/jobs/ingest$"),
+        body=to_body(
+            {
+                "apiVersion": 52.0,
+                "columnDelimiter": "COMMA",
+                "concurrencyMode": "Parallel",
+                "contentType": "CSV",
+                "id": "Job-1",
+                "jobType": "V2Ingest",
+                "lineEnding": "LF",
+                "object": "Contact",
+                "operation": operation,
+                "state": JobState.open,
+            }
+        ),
+        status=http.OK,
+    )
+    responses.add(
+        responses.PUT,
+        re.compile(r"^https://.*/jobs/ingest/Job-1/batches$"),
+        body="",
+        status=http.CREATED,
+    )
+    responses.add(
+        responses.PATCH,
+        re.compile(r"^https://.*/jobs/ingest/Job-1$"),
+        body=to_body(
+            {
+                "apiVersion": 52.0,
+                "columnDelimiter": "COMMA",
+                "concurrencyMode": "Parallel",
+                "contentType": "CSV",
+                "id": "Job-1",
+                "jobType": "V2Ingest",
+                "lineEnding": "LF",
+                "object": "Contact",
+                "operation": operation,
+                "state": JobState.upload_complete,
+            }
+        ),
+        status=http.OK,
+    )
+    responses.add(
+        responses.GET,
+        re.compile(r"^https://.*/jobs/ingest/Job-1$"),
+        body=to_body(
+            {
+                "apiVersion": 52.0,
+                "columnDelimiter": "COMMA",
+                "concurrencyMode": "Parallel",
+                "contentType": "CSV",
+                "id": "Job-1",
+                "jobType": "V2Ingest",
+                "lineEnding": "LF",
+                "object": "Contact",
+                "operation": operation,
+                "state": JobState.in_progress,
+            }
+        ),
+        status=http.OK,
+    )
+    responses.add(
+        responses.GET,
+        re.compile(r"^https://.*/jobs/ingest/Job-1$"),
+        body=to_body(
+            {
+                "apiVersion": 52.0,
+                "columnDelimiter": "COMMA",
+                "concurrencyMode": "Parallel",
+                "contentType": "CSV",
+                "id": "Job-1",
+                "jobType": "V2Ingest",
+                "lineEnding": "LF",
+                "numberRecordsFailed": failed,
+                "numberRecordsProcessed": processed,
+                "object": "Contact",
+                "operation": operation,
+                "state": JobState.job_complete,
+            }
+        ),
+    )
+
+
+def ingest_data(operation, data, **kwargs):
+    """Upload data into Salesforce"""
+    operation = (
+        "hard_delete" if operation == Operation.hard_delete else operation
+    )
+    with to_csv_file(data) as csv_file:
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        results = getattr(client.bulk2.Contact, operation)(csv_file, **kwargs)
+    return results
+
+
+class TestSFBulk2Type(unittest.TestCase):
+    """Test SFBulk2Type"""
+
+    def setUp(self):
+        request_patcher = patch("simple_salesforce.api.requests")
+        self.mockrequest = request_patcher.start()
+        self.addCleanup(request_patcher.stop)
+        self.expected_results = [
+            {
+                "numberRecordsFailed": 0,
+                "numberRecordsProcessed": 2,
+                "numberRecordsTotal": 2,
+                "job_id": "Job-1",
+            }
+        ]
+        self.expected_query = [
+            textwrap.dedent(
+                """
+                "Id","AccountId","Email","FirstName","LastName"
+                "001xx000003DHP0AAO","ID-13","contact1@example.com","Bob","x"
+                "001xx000003DHP1AAO","ID-24","contact2@example.com","Alice","y"
+                "001xx000003DHP0AAO","ID-13","contact1@example.com","Bob","x"
+                "001xx000003DHP1AAO","ID-24","contact2@example.com","Alice","y"
+                """
+            )
+        ]
+        self.insert_data = [
+            {
+                "Custom_Id__c": "CustomID1",
+                "AccountId": "ID-13",
+                "Email": "contact1@example.com",
+                "FirstName": "Bob",
+                "LastName": "x",
+            },
+            {
+                "Custom_Id__c": "CustomID2",
+                "AccountId": "ID-24",
+                "Email": "contact2@example.com",
+                "FirstName": "Alice",
+                "LastName": "y",
+            },
+        ]
+        self.upsert_data = [
+            {
+                "Custom_Id__c": "CustomID1",
+                "LastName": "X",
+            },
+            {
+                "Custom_Id__c": "CustomID2",
+                "LastName": "Y",
+            },
+        ]
+        self.delete_data = [
+            {"Id": "a011s00000DTU9zAAH"},
+            {"Id": "a011s00000DTUA0AAP"},
+        ]
+
+    def test_bulk2_type(self):
+        """Test bulk2 type creation"""
+        session = requests.Session()
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=session,
+        )
+        contact = client.bulk2.Contact
+        self.assertIs("Contact", contact.object_name)
+        self.assertIs(client.bulk2_url, contact.bulk2_url)
+        self.assertEqual(tests.BULKv2_HEADERS, contact.headers)
+
+    @responses.activate
+    def test_insert(self):
+        """Test bulk2 insert records"""
+        operation = Operation.insert
+        total = len(self.insert_data)
+        ingest_responses(operation, processed=total, failed=0)
+        results = ingest_data(operation, self.insert_data)
+        self.assertEqual(self.expected_results, results)
+
+        responses.add(
+            responses.GET,
+            re.compile(r"^https://.*/jobs/ingest/Job-1/successfulResults$"),
+            body=textwrap.dedent(
+                """
+                "sf__Id","sf__Created","Custom_Id__c","AccountId","Email","FirstName","LastName"
+                "a011s00000DML8XAAX","true","CustomID1","ID-13","contact1@example.com","Bob","x"
+                "a011s00000DML8YAAX","true","CustomID2","ID-24","contact2@example.com","Alice","y"
+                """
+            ),
+            status=http.OK,
+        )
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        results = client.bulk2.Contact.get_successful_records("Job-1")
+        self.assertEqual(
+            textwrap.dedent(
+                """
+                "sf__Id","sf__Created","Custom_Id__c","AccountId","Email","FirstName","LastName"
+                "a011s00000DML8XAAX","true","CustomID1","ID-13","contact1@example.com","Bob","x"
+                "a011s00000DML8YAAX","true","CustomID2","ID-24","contact2@example.com","Alice","y"
+            """
+            ),
+            results,
+        )
+
+    @responses.activate
+    def test_upsert(self):
+        """Test bulk2 upsert records"""
+        operation = Operation.upsert
+        total = len(self.upsert_data)
+        ingest_responses(operation, processed=total, failed=0)
+        results = ingest_data(
+            operation, self.upsert_data, external_id_field="Custom_Id__c"
+        )
+        self.assertEqual(self.expected_results, results)
+
+        responses.add(
+            responses.GET,
+            re.compile(r"^https://.*/jobs/ingest/Job-1/successfulResults$"),
+            body=textwrap.dedent(
+                """
+                "sf__Id","sf__Created","Custom_Id__c","LastName"
+                "a011s00000DML8XAAX","false","CustomID1","X"
+                "a011s00000DML8YAAX","false","CustomID2","Y"
+                """
+            ),
+            status=http.OK,
+        )
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        results = client.bulk2.Contact.get_successful_records("Job-1")
+        self.assertEqual(
+            textwrap.dedent(
+                """
+                 "sf__Id","sf__Created","Custom_Id__c","LastName"
+                 "a011s00000DML8XAAX","false","CustomID1","X"
+                 "a011s00000DML8YAAX","false","CustomID2","Y"
+                """
+            ),
+            results,
+        )
+
+    @responses.activate
+    def test_hard_delete(self):
+        """Test bulk2 hardDelete records"""
+        operation = Operation.hard_delete
+        total = len(self.delete_data)
+        ingest_responses(operation, processed=total, failed=0)
+        results = ingest_data(operation, self.delete_data)
+        self.assertEqual(self.expected_results, results)
+
+        responses.add(
+            responses.GET,
+            re.compile(r"^https://.*/jobs/ingest/Job-1/successfulResults$"),
+            body=textwrap.dedent(
+                """
+                "sf__Id","sf__Created","Id"
+                "a011s00000DTU9zAAH","false","a011s00000DTU9zAAH"
+                "a011s00000DTUA0AAP","false","a011s00000DTUA0AAP"
+                """
+            ),
+            status=http.OK,
+        )
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        results = client.bulk2.Contact.get_successful_records("Job-1")
+        self.assertEqual(
+            textwrap.dedent(
+                """
+                "sf__Id","sf__Created","Id"
+                "a011s00000DTU9zAAH","false","a011s00000DTU9zAAH"
+                "a011s00000DTUA0AAP","false","a011s00000DTUA0AAP"
+                """
+            ),
+            results,
+        )
+
+    @responses.activate
+    def test_query(self):
+        """Test bulk2 query records"""
+        operation = Operation.query
+        responses.add(
+            responses.POST,
+            re.compile(r"^https://.*/jobs/query$"),
+            body=to_body(
+                {
+                    "apiVersion": 52.0,
+                    "columnDelimiter": "COMMA",
+                    "concurrencyMode": "Parallel",
+                    "contentType": "CSV",
+                    "id": "Job-1",
+                    "lineEnding": "LF",
+                    "object": "Contact",
+                    "operation": operation,
+                    "state": JobState.upload_complete,
+                }
+            ),
+            status=http.OK,
+        )
+        responses.add(
+            responses.GET,
+            re.compile(r"^https://.*/jobs/query/Job-1$"),
+            body=to_body(
+                {
+                    "apiVersion": 52.0,
+                    "columnDelimiter": "COMMA",
+                    "concurrencyMode": "Parallel",
+                    "contentType": "CSV",
+                    "id": "Job-1",
+                    "jobType": "V2Query",
+                    "lineEnding": "LF",
+                    "numberRecordsProcessed": 4,
+                    "object": "Contact",
+                    "operation": operation,
+                    "state": JobState.in_progress,
+                }
+            ),
+            status=http.OK,
+        )
+        responses.add(
+            responses.GET,
+            re.compile(r"^https://.*/jobs/query/Job-1$"),
+            body=to_body(
+                {
+                    "apiVersion": 52.0,
+                    "columnDelimiter": "COMMA",
+                    "concurrencyMode": "Parallel",
+                    "contentType": "CSV",
+                    "id": "Job-1",
+                    "jobType": "V2Query",
+                    "lineEnding": "LF",
+                    "numberRecordsProcessed": 4,
+                    "object": "Contact",
+                    "operation": operation,
+                    "state": JobState.job_complete,
+                }
+            ),
+            status=http.OK,
+        )
+        responses.add(
+            responses.GET,
+            re.compile(
+                r"^https://.*/jobs/query/Job-1/results\?maxRecords=\d+$"
+            ),
+            body=textwrap.dedent(
+                """
+                "Id","AccountId","Email","FirstName","LastName"
+                "001xx000003DHP0AAO","ID-13","contact1@example.com","Bob","x"
+                "001xx000003DHP1AAO","ID-24","contact2@example.com","Alice","y"
+                "001xx000003DHP0AAO","ID-13","contact1@example.com","Bob","x"
+                "001xx000003DHP1AAO","ID-24","contact2@example.com","Alice","y"
+                """
+            ),
+            headers={
+                "Sforce-NumberOfRecords": "4",
+                "Sforce-Query-Locator": "",
+            },
+            status=http.OK,
+        )
+
+        query = "SELECT Id,AccountId,Email,FirstName,LastName FROM Contact"
+        session = requests.Session()
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=session,
+        )
+        results = client.bulk2.Contact.query(query)
+        results = list(results)
+        self.assertEqual(self.expected_query, results)
+
+    @responses.activate
+    def test_get_failed_record_results(self):
+        """Test bulk2 get failed records"""
+        operation = Operation.insert
+        total = len(self.insert_data)
+        ingest_responses(
+            operation,
+            processed=total,
+            failed=total,
+        )
+        results = ingest_data(operation, self.insert_data)
+        self.assertEqual(
+            [
+                {
+                    "numberRecordsFailed": total,
+                    "numberRecordsProcessed": total,
+                    "numberRecordsTotal": total,
+                    "job_id": "Job-1",
+                }
+            ],
+            results,
+        )
+
+        def make_responses():
+            responses.add(
+                responses.GET,
+                re.compile(r"^https://.*/jobs/ingest/Job-1/failedResults$"),
+                body=textwrap.dedent(
+                    """
+                        "sf__Id","sf__Error","Custom_Id__c","AccountId","Email","FirstName","LastName"
+                        "","UNABLE_TO_LOCK_ROW","CustomID1","ID-13","contact1@example.com","Bob","x"
+                        "","UNABLE_TO_LOCK_ROW","CustomID2","ID-24","contact2@example.com","Alice","y"
+                        """
+                ),
+                status=http.OK,
+            )
+
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        expected_results = textwrap.dedent(
+            """
+                "sf__Id","sf__Error","Custom_Id__c","AccountId","Email","FirstName","LastName"
+                "","UNABLE_TO_LOCK_ROW","CustomID1","ID-13","contact1@example.com","Bob","x"
+                "","UNABLE_TO_LOCK_ROW","CustomID2","ID-24","contact2@example.com","Alice","y"
+                """
+        )
+
+        make_responses()
+        failed_results = client.bulk2.Contact.get_failed_records("Job-1")
+        self.assertEqual(expected_results, failed_results)
+
+        make_responses()
+        with temp_csv_file() as csv_file:
+            client.bulk2.Contact.get_failed_records("Job-1", file=csv_file)
+            with open(csv_file, "r", encoding="utf-8") as bis:
+                self.assertEqual(expected_results, bis.read())
+
+    @responses.activate
+    def test_get_unprocessed_record_results(self):
+        """Test bulk2 get unprocessed records"""
+        operation = Operation.insert
+        total = len(self.insert_data)
+        ingest_responses(
+            operation,
+            processed=total - 1,
+            failed=0,
+        )
+        results = ingest_data(operation, self.insert_data)
+        self.assertEqual(
+            [
+                {
+                    "numberRecordsFailed": 0,
+                    "numberRecordsProcessed": total - 1,
+                    "numberRecordsTotal": total,
+                    "job_id": "Job-1",
+                }
+            ],
+            results,
+        )
+
+        def make_responses():
+            responses.add(
+                responses.GET,
+                re.compile(
+                    r"^https://.*/jobs/ingest/Job-1/unprocessedRecords$"
+                ),
+                body=textwrap.dedent(
+                    """
+                        "Custom_Id__c","AccountId","Email","FirstName","LastName"
+                        "CustomID2","ID-24","contact2@example.com","Alice","y"
+                        """
+                ),
+                status=http.OK,
+            )
+
+        client = Salesforce(
+            session_id=tests.SESSION_ID,
+            instance_url=tests.SERVER_URL,
+            session=requests.Session(),
+        )
+        expected_results = textwrap.dedent(
+            """
+            "Custom_Id__c","AccountId","Email","FirstName","LastName"
+            "CustomID2","ID-24","contact2@example.com","Alice","y"
+            """
+        )
+
+        make_responses()
+        results = client.bulk2.Contact.get_unprocessed_records("Job-1")
+        self.assertEqual(expected_results, results)
+
+        make_responses()
+        with temp_csv_file() as csv_file:
+            client.bulk2.Contact.get_unprocessed_records("Job-1", file=csv_file)
+            with open(csv_file, "r", encoding="utf-8") as bis:
+                self.assertEqual(expected_results, bis.read())
diff --git a/simple_salesforce/tests/test_login.py b/simple_salesforce/tests/test_login.py
index 70d65088..0c18201c 100644
--- a/simple_salesforce/tests/test_login.py
+++ b/simple_salesforce/tests/test_login.py
@@ -1,10 +1,10 @@
 """Tests for login.py"""
 
 import http.client as http
-import os
 import re
 import unittest
 import warnings
+from pathlib import Path
 from unittest.mock import Mock, patch
 from urllib.parse import urlparse
 
@@ -133,10 +133,10 @@ def test_failure(self):
     @responses.activate
     def test_token_login_success_with_key_file(self):
         """Test a successful JWT Token login with a key file"""
-        pkey_file = os.path.join(os.path.dirname(__file__), 'sample-key.pem')
+        pkey_path = Path(__file__).parent / 'sample-key.pem'
         login_args = {
             'consumer_key': '12345.abcde',
-            'privatekey_file': pkey_file
+            'privatekey_file': str(pkey_path)
             }
         self._test_login_success(
             re.compile(r'^https://login.salesforce.com/.*$'), login_args,
@@ -145,9 +145,8 @@ def test_token_login_success_with_key_file(self):
     @responses.activate
     def test_token_login_success_with_key(self):
         """Test a successful JWT Token login with a key from a string"""
-        pkey_file = os.path.join(os.path.dirname(__file__), 'sample-key.pem')
-        with open(pkey_file, 'rb') as key_file:
-            key = key_file.read().decode("utf-8")
+        pkey_path = Path(__file__).parent / 'sample-key.pem'
+        key = pkey_path.read_bytes().decode()
 
         login_args = {
             'consumer_key': '12345.abcde',
@@ -160,9 +159,8 @@ def test_token_login_success_with_key(self):
     @responses.activate
     def test_token_login_success_with_key_bytes(self):
         """Test a successful JWT Token login with key bytes"""
-        pkey_file = os.path.join(os.path.dirname(__file__), 'sample-key.pem')
-        with open(pkey_file, 'rb') as key:
-            key_bytes = key.read()
+        pkey_path = Path(__file__).parent / 'sample-key.pem'
+        key_bytes = pkey_path.read_bytes()
 
         login_args = {
             'consumer_key': '12345.abcde',
@@ -186,9 +184,7 @@ def test_token_login_failure(self):
             SalesforceLogin(
                 username='myemail@example.com.sandbox',
                 consumer_key='12345.abcde',
-                privatekey_file=os.path.join(
-                    os.path.dirname(__file__), 'sample-key.pem'
-                    )
+                privatekey_file=str(Path(__file__).parent / 'sample-key.pem')
                 )
         self.assertTrue(self.mockrequest.post.called)
 
@@ -222,8 +218,8 @@ def on_response(*args, **kwargs):
                     session=session,
                     username='foo@bar.com',
                     consumer_key='12345.abcde',
-                    privatekey_file=os.path.join(
-                        os.path.dirname(__file__), 'sample-key.pem'
+                    privatekey_file=str(
+                        Path(__file__).parent / 'sample-key.pem'
                         )
                     )
             assert len(warning) >= 1
@@ -235,6 +231,7 @@ def on_response(*args, **kwargs):
     def test_connected_app_login_success(self):
         """Test a successful connected app login with a key file"""
         login_args = {
+            'password': 'password',
             'consumer_key': '12345.abcde',
             'consumer_secret': '12345.abcde'
             }
@@ -255,6 +252,7 @@ def test_connected_app_login_failure(self):
         with self.assertRaises(SalesforceAuthenticationFailed):
             SalesforceLogin(
                 username='myemail@example.com.sandbox',
+                password='password',
                 consumer_key='12345.abcde',
                 consumer_secret='12345.abcde'
                 )