Py 3.12 Changes

bin/cwutil

@@ -7,7 +7,7 @@
 import boto
 cw = boto.connect_cloudwatch()
 
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 
 def _parse_time(time_string):
     """Internal function to parse a time string"""
@@ -52,11 +52,11 @@ def stats(namespace, metric_name, dimensions=None, statistics="Average", start_t
     if end_time:
         end_time = _parse_time(end_time)
     else:
-        end_time = datetime.utcnow()
+        end_time = datetime.now(tz=timezone.utc).replace(tzinfo=None)
     if start_time:
         start_time = _parse_time(start_time)
     else:
-        start_time = datetime.utcnow() - timedelta(days=1)
+        start_time = datetime.now(tz=timezone.utc).replace(tzinfo=None) - timedelta(days=1)
 
     print "%-30s %s" % ('Timestamp', statistics)
     print "-"*50

boto/__init__.py

@@ -1136,7 +1136,7 @@ def storage_uri(uri_str, default_scheme='file', debug=0, validate=True,
     * gs://bucket
     * s3://bucket
     * filename (which could be a Unix path like /a/b/c or a Windows path like
-      C:\a\b\c)
+      C:\a\b\\c)
 
     The last example uses the default scheme ('file', unless overridden).
     """

boto/auth.py

@@ -547,7 +547,7 @@ def add_auth(self, req, **kwargs):
         # authorization header is removed first.
         if 'X-Amzn-Authorization' in req.headers:
             del req.headers['X-Amzn-Authorization']
-        now = datetime.datetime.utcnow()
+        now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
         req.headers['X-Amz-Date'] = now.strftime('%Y%m%dT%H%M%SZ')
         if self._provider.security_token:
             req.headers['X-Amz-Security-Token'] = self._provider.security_token
@@ -777,7 +777,7 @@ def presign(self, req, expires, iso_date=None):
         http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-query-string-auth.html
         """
         if iso_date is None:
-            iso_date = datetime.datetime.utcnow().strftime('%Y%m%dT%H%M%SZ')
+            iso_date = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None).strftime('%Y%m%dT%H%M%SZ')
 
         region = self.determine_region_name(req.host)
         service = self.determine_service_name(req.host)

boto/cloudfront/distribution.py

@@ -540,7 +540,7 @@ def create_signed_url(self, url, keypair_id,
         :type policy_url: str
         :param policy_url: If provided, allows the signature to contain
             wildcard globs in the URL.  For example, you could
-            provide: 'http://example.com/media/\*' and the policy
+            provide: 'http://example.com/media/*' and the policy
             and signature would allow access to all contents of
             the media subdirectory. If not specified, only
             allow access to the exact url provided in 'url'.

boto/cloudsearchdomain/layer1.py

@@ -270,7 +270,7 @@ def search(self, query, cursor=None, expr=None, facet=None,
               ~ operator to perform a sloppy phrase search. Disabling the `fuzzy`
               operator disables the ability to use the ~ operator to perform a
               fuzzy search. `escape` disables the ability to use a backslash (
-              `\`) to escape special characters within the search string.
+              `\\`) to escape special characters within the search string.
               Disabling whitespace is an advanced option that prevents the parser
               from tokenizing on whitespace, which can be useful for Vietnamese.
               (It prevents Vietnamese words from being split incorrectly.) For

boto/connection.py

@@ -668,10 +668,10 @@ def handle_proxy(self, proxy, proxy_port, proxy_user, proxy_pass):
         self.proxy_pass = proxy_pass
         if 'http_proxy' in os.environ and not self.proxy:
             pattern = re.compile(
-                '(?:http://)?'
-                '(?:(?P<user>[\w\-\.]+):(?P<pass>.*)@)?'
-                '(?P<host>[\w\-\.]+)'
-                '(?::(?P<port>\d+))?'
+                r'(?:http://)?'
+                r'(?:(?P<user>[\w\-\.]+):(?P<pass>.*)@)?'
+                r'(?P<host>[\w\-\.]+)'
+                r'(?::(?P<port>\d+))?'
             )
             match = pattern.match(os.environ['http_proxy'])
             if match:
@@ -1117,8 +1117,8 @@ def _get_correct_s3_endpoint_from_response(self, request,
             return self._fix_s3_endpoint_region(request.host, err.region)
         elif err.error_code == 'IllegalLocationConstraintException':
             region_regex = (
-                'The (.*?) location constraint is incompatible for the region '
-                'specific endpoint this request was sent to\.'
+                r'The (.*?) location constraint is incompatible for the region '
+                r'specific endpoint this request was sent to\.'
             )
             match = re.search(region_regex, err.body)
             if match and match.group(1) != 'unspecified':

boto/ec2/connection.py

@@ -29,6 +29,7 @@
 import warnings
 from datetime import datetime
 from datetime import timedelta
+from datetime import timezone
 
 import boto
 from boto.auth import detect_potential_sigv4
@@ -2600,7 +2601,7 @@ def trim_snapshots(self, hourly_backups=8, daily_backups=7,
         # target time (we delete a snapshot if there's another snapshot
         # that was made closer to the preceding target time).
 
-        now = datetime.utcnow()
+        now = datetime.now(tz=timezone.utc).replace(tzinfo=None)
         last_hour = datetime(now.year, now.month, now.day, now.hour)
         last_midnight = datetime(now.year, now.month, now.day)
         last_sunday = datetime(now.year, now.month, now.day) - timedelta(days=(now.weekday() + 1) % 7)

boto/gs/resumable_upload_handler.py

@@ -237,7 +237,7 @@ def _query_server_pos(self, conn, file_length):
         range_spec = resp.getheader('range')
         if range_spec:
             # Parse 'bytes=<from>-<to>' range_spec.
-            m = re.search('bytes=(\d+)-(\d+)', range_spec)
+            m = re.search(r'bytes=(\d+)-(\d+)', range_spec)
             if m:
                 server_start = long(m.group(1))
                 server_end = long(m.group(2))

boto/https_connection.py

@@ -77,7 +77,7 @@ def ValidateCertificateHostname(cert, hostname):
         "validating server certificate: hostname=%s, certificate hosts=%s",
         hostname, hosts)
     for host in hosts:
-        host_re = host.replace('.', '\.').replace('*', '[^.]*')
+        host_re = host.replace(r'.', r'\.').replace(r'*', r'[^.]*')
         if re.search('^%s$' % (host_re,), hostname, re.I):
             return True
     return False

boto/manage/task.py

@@ -63,7 +63,7 @@ def __init__(self, id=None, **kw):
         super(Task, self).__init__(id, **kw)
         self.hourly = self.hour == '*'
         self.daily = self.hour != '*'
-        self.now = datetime.datetime.utcnow()
+        self.now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
 
     def check(self):
         """

boto/mturk/connection.py

@@ -919,7 +919,7 @@ def _has_expired(self):
         """ Has this HIT expired yet? """
         expired = False
         if hasattr(self, 'Expiration'):
-            now = datetime.datetime.utcnow()
+            now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
             expiration = datetime.datetime.strptime(self.Expiration, '%Y-%m-%dT%H:%M:%SZ')
             expired = (now >= expiration)
         else:

boto/opsworks/layer1.py

@@ -2657,7 +2657,7 @@ def update_layer(self, layer_id, name=None, shortname=None,
             OpsWorksand by Chef. The short name is also used as the name for
             the directory where your app files are installed. It can have a
             maximum of 200 characters and must be in the following format:
-            /\A[a-z0-9\-\_\.]+\Z/.
+            /\\A[a-z0-9\\-\\_\\.]+\\Z/.
 
         :type attributes: map
         :param attributes: One or more user-defined key/value pairs to be added

boto/provider.py

@@ -29,6 +29,7 @@
 import os
 from boto.compat import six
 from datetime import datetime
+from datetime import timezone
 
 import boto
 from boto import config
@@ -253,7 +254,7 @@ def _credentials_need_refresh(self):
         else:
             # The credentials should be refreshed if they're going to expire
             # in less than 5 minutes.
-            delta = self._credential_expiry_time - datetime.utcnow()
+            delta = self._credential_expiry_time - datetime.now(tz=timezone.utc).replace(tzinfo=None)
             # python2.6 does not have timedelta.total_seconds() so we have
             # to calculate this ourselves.  This is straight from the
             # datetime docs.

boto/pyami/config.py

@@ -95,7 +95,7 @@ def load_credential_file(self, path):
     def load_from_path(self, path):
         file = open(path)
         for line in file.readlines():
-            match = re.match("^#import[\s\t]*([^\s^\t]*)[\s\t]*$", line)
+            match = re.match(r"^#import[\s\t]*([^\s^\t]*)[\s\t]*$", line)
             if match:
                 extended_file = match.group(1)
                 (dir, file) = os.path.split(path)

boto/redshift/layer1.py

@@ -417,7 +417,7 @@ def create_cluster(self, cluster_identifier, node_type, master_username,
         + Must contain at least one lowercase letter.
         + Must contain one number.
         + Can be any printable ASCII character (ASCII code 33 to 126) except '
-              (single quote), " (double quote), \, /, @, or space.
+              (single quote), " (double quote), \\, /, @, or space.
 
         :type cluster_security_groups: list
         :param cluster_security_groups: A list of security groups to be
@@ -2189,7 +2189,7 @@ def enable_logging(self, cluster_identifier, bucket_name,
 
         + Cannot exceed 512 characters
         + Cannot contain spaces( ), double quotes ("), single quotes ('), a
-              backslash (\), or control characters. The hexadecimal codes for
+              backslash (\\), or control characters. The hexadecimal codes for
               invalid characters are:
 
             + x00 to x20
@@ -2363,7 +2363,7 @@ def modify_cluster(self, cluster_identifier, cluster_type=None,
         + Must contain at least one lowercase letter.
         + Must contain one number.
         + Can be any printable ASCII character (ASCII code 33 to 126) except '
-              (single quote), " (double quote), \, /, @, or space.
+              (single quote), " (double quote), \\, /, @, or space.
 
         :type cluster_parameter_group_name: string
         :param cluster_parameter_group_name: The name of the cluster parameter

boto/sdb/db/manager/sdbmanager.py

@@ -345,7 +345,7 @@ def encode_blob(self, value):
             key = bucket.new_key(str(uuid.uuid4()))
             value.id = "s3://%s/%s" % (key.bucket.name, key.name)
         else:
-            match = re.match("^s3:\/\/([^\/]*)\/(.*)$", value.id)
+            match = re.match(r"^s3://([^/]*)/(.*)$", value.id)
             if match:
                 s3 = self.manager.get_s3_connection()
                 bucket = s3.get_bucket(match.group(1), validate=False)
@@ -360,7 +360,7 @@ def encode_blob(self, value):
     def decode_blob(self, value):
         if not value:
             return None
-        match = re.match("^s3:\/\/([^\/]*)\/(.*)$", value)
+        match = re.match(r"^s3://([^/]*)/(.*)$", value)
         if match:
             s3 = self.manager.get_s3_connection()
             bucket = s3.get_bucket(match.group(1), validate=False)

boto/sdb/db/property.py

@@ -262,7 +262,7 @@ class S3KeyProperty(Property):
 
     data_type = boto.s3.key.Key
     type_name = 'S3Key'
-    validate_regex = "^s3:\/\/([^\/]*)\/(.*)$"
+    validate_regex = r"^s3://([^/]*)/(.*)$"
 
     def __init__(self, verbose_name=None, name=None, default=None,
                  required=False, validator=None, choices=None, unique=False):
@@ -421,7 +421,7 @@ def get_value_for_datastore(self, model_instance):
         return super(DateTimeProperty, self).get_value_for_datastore(model_instance)
 
     def now(self):
-        return datetime.datetime.utcnow()
+        return datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
 
 
 class DateProperty(Property):

boto/sts/connection.py

@@ -115,7 +115,7 @@ def _required_auth_capability(self):
     def _check_token_cache(self, token_key, duration=None, window_seconds=60):
         token = _session_token_cache.get(token_key, None)
         if token:
-            now = datetime.datetime.utcnow()
+            now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
             expires = boto.utils.parse_ts(token.expiration)
             delta = expires - now
             if delta < datetime.timedelta(seconds=window_seconds):

boto/sts/credentials.py

@@ -132,7 +132,7 @@ def is_expired(self, time_offset_seconds=0):
         :param time_offset_seconds: The number of seconds into the future
             to test the Session Token for expiration.
         """
-        now = datetime.datetime.utcnow()
+        now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
         if time_offset_seconds:
             now = now + datetime.timedelta(seconds=time_offset_seconds)
         ts = boto.utils.parse_ts(self.expiration)

docs/source/cloudsearch_tut.rst

@@ -423,10 +423,11 @@ It is also possible to delete documents::
 
     >>> import time
     >>> from datetime import datetime
+    >>> from datetime import timezone
 
     >>> doc_service = domain.get_document_service()
 
     >>> # Again we'll cheat and use the current epoch time as our version number
 
-    >>> doc_service.delete(4, int(time.mktime(datetime.utcnow().timetuple())))
+    >>> doc_service.delete(4, int(time.mktime(datetime.now(tz=timezone.utc).replace(tzinfo=None).timetuple())))
     >>> doc_service.commit()

docs/source/cloudwatch_tut.rst

@@ -79,7 +79,7 @@ that we are interested in.  For this example, let's say we want the
 data for the previous hour::
 
     >>> import datetime
-    >>> end = datetime.datetime.utcnow()
+    >>> end = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
     >>> start = end - datetime.timedelta(hours=1)
 
 We also need to supply the Statistic that we want reported and

tests/db/test_password.py

@@ -81,20 +81,20 @@ class MyModel(Model):
         id= obj.id
         time.sleep(5)
         obj = MyModel.get_by_id(id)
-        self.assertEquals(obj.password, 'bar')
-        self.assertEquals(str(obj.password), expected)
+        self.assertEqual(obj.password, 'bar')
+        self.assertEqual(str(obj.password), expected)
                           #hmac.new('mysecret','bar').hexdigest())
 
 
     def test_aaa_default_password_property(self):
         cls = self.test_model()
         obj = cls(id='passwordtest')
         obj.password = 'foo'
-        self.assertEquals('foo', obj.password)
+        self.assertEqual('foo', obj.password)
         obj.save()
         time.sleep(5)
         obj = cls.get_by_id('passwordtest')
-        self.assertEquals('foo', obj.password)
+        self.assertEqual('foo', obj.password)
 
     def test_password_constructor_hashfunc(self):
         import hmac
@@ -103,8 +103,8 @@ def test_password_constructor_hashfunc(self):
         obj = cls()
         obj.password='hello'
         expected = myhashfunc('hello').hexdigest()
-        self.assertEquals(obj.password, 'hello')
-        self.assertEquals(str(obj.password), expected)
+        self.assertEqual(obj.password, 'hello')
+        self.assertEqual(str(obj.password), expected)
         obj.save()
         id = obj.id
         time.sleep(5)

tests/integration/cloudformation/test_connection.py

@@ -136,7 +136,7 @@ def test_large_template_stack_size(self):
         self.assertEqual(self.stack_name, stack.stack_name)
 
         params = [(p.key, p.value) for p in stack.parameters]
-        self.assertEquals([('Parameter1', 'initial_value'),
+        self.assertEqual([('Parameter1', 'initial_value'),
                            ('Parameter2', 'initial_value')], params)
 
         for _ in range(30):
@@ -154,7 +154,7 @@ def test_large_template_stack_size(self):
         stacks = self.connection.describe_stacks(self.stack_name)
         stack = stacks[0]
         params = [(p.key, p.value) for p in stacks[0].parameters]
-        self.assertEquals([('Parameter1', 'initial_value'),
+        self.assertEqual([('Parameter1', 'initial_value'),
                            ('Parameter2', 'updated_value')], params)
 
         # Waiting for the update to complete to unblock the delete_stack in the

tests/integration/ec2/cloudwatch/test_connection.py

@@ -220,14 +220,14 @@ def test_build_put_params_invalid(self):
     def test_get_metric_statistics(self):
         c = CloudWatchConnection()
         m = c.list_metrics()[0]
-        end = datetime.datetime.utcnow()
+        end = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
         start = end - datetime.timedelta(hours=24 * 14)
         c.get_metric_statistics(
             3600 * 24, start, end, m.name, m.namespace, ['Average', 'Sum'])
 
     def test_put_metric_data(self):
         c = CloudWatchConnection()
-        now = datetime.datetime.utcnow()
+        now = datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None)
         name, namespace = 'unit-test-metric', 'boto-unit-test'
         c.put_metric_data(namespace, name, 5, now, 'Bytes')
 
@@ -240,7 +240,7 @@ def test_put_metric_data(self):
         # time.sleep(60)
         # l = metric.query(
         #     now - datetime.timedelta(seconds=60),
-        #     datetime.datetime.utcnow(),
+        #     datetime.datetime.now(tz=datetime.timezone.utc).replace(tzinfo=None),
         #     'Average')
         # assert l
         # for row in l:
@@ -261,13 +261,13 @@ def read(self):
 
         c.make_request = make_request
         alarms = c.describe_alarms()
-        self.assertEquals(alarms.next_token, 'mynexttoken')
-        self.assertEquals(alarms[0].name, 'FancyAlarm')
-        self.assertEquals(alarms[0].comparison, '<')
-        self.assertEquals(alarms[0].dimensions, {u'Job': [u'ANiceCronJob']})
-        self.assertEquals(alarms[1].name, 'SuperFancyAlarm')
-        self.assertEquals(alarms[1].comparison, '>')
-        self.assertEquals(alarms[1].dimensions, {u'Job': [u'ABadCronJob']})
+        self.assertEqual(alarms.next_token, 'mynexttoken')
+        self.assertEqual(alarms[0].name, 'FancyAlarm')
+        self.assertEqual(alarms[0].comparison, '<')
+        self.assertEqual(alarms[0].dimensions, {u'Job': [u'ANiceCronJob']})
+        self.assertEqual(alarms[1].name, 'SuperFancyAlarm')
+        self.assertEqual(alarms[1].comparison, '>')
+        self.assertEqual(alarms[1].dimensions, {u'Job': [u'ABadCronJob']})
 
 if __name__ == '__main__':
     unittest.main()