From: Harmon Date: Mon, 28 Dec 2020 04:49:52 +0000 (-0600) Subject: Replace old C-style string formatting/interpolation with f-strings X-Git-Url: https://vcs.fsf.org/?a=commitdiff_plain;h=d036c38671326113efa67bfeac7dba955733cd44;p=tweepy.git Replace old C-style string formatting/interpolation with f-strings --- diff --git a/setup.py b/setup.py index 77e2ded..07fab6f 100644 --- a/setup.py +++ b/setup.py @@ -11,7 +11,7 @@ with open(VERSION_FILE) as version_file: if match: version = match.group(1) else: - raise RuntimeError("Unable to find version string in %s." % (VERSION_FILE,)) + raise RuntimeError(f"Unable to find version string in {VERSION_FILE}.") with open("README.md") as readme_file: long_description = readme_file.read() diff --git a/tests/test_api.py b/tests/test_api.py index cf1f6cb..f86bbcf 100644 --- a/tests/test_api.py +++ b/tests/test_api.py @@ -428,7 +428,7 @@ class TweepyCacheTests(unittest.TestCase): # test count for i in range(0, 20): - self.cache.store('testkey%i' % i, 'testvalue') + self.cache.store(f'testkey{i}', 'testvalue') self.assertEqual(self.cache.count(), 20, 'Count is wrong') # test flush diff --git a/tests/test_auth.py b/tests/test_auth.py index d82ff73..c1dfb00 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -20,7 +20,7 @@ class TweepyAuthTests(unittest.TestCase): # build api object test using oauth api = API(auth) - s = api.update_status('test %i' % random.randint(0, 1000)) + s = api.update_status(f'test {random.randint(0, 1000)}') api.destroy_status(s.id) def testaccesstype(self): diff --git a/tests/test_streaming.py b/tests/test_streaming.py index c64b516..592a4cc 100644 --- a/tests/test_streaming.py +++ b/tests/test_streaming.py @@ -29,7 +29,7 @@ class MockStreamListener(StreamListener): return False def on_error(self, code): - print("response: %s" % code) + print(f"response: {code}") return True def on_status(self, status): diff --git a/tweepy/api.py b/tweepy/api.py index 7276d83..e514fc6 100644 --- a/tweepy/api.py +++ b/tweepy/api.py @@ -1381,18 +1381,16 @@ class API: if f is None: try: if os.path.getsize(filename) > (max_size * 1024): - raise TweepError('File is too big, must be less than %skb.' - % max_size) + raise TweepError(f'File is too big, must be less than {max_size}kb.') except os.error as e: - raise TweepError('Unable to access file: %s' % e.strerror) + raise TweepError(f'Unable to access file: {e.strerror}') # build the mulitpart-formdata body fp = open(filename, 'rb') else: f.seek(0, 2) # Seek to end of file if f.tell() > (max_size * 1024): - raise TweepError('File is too big, must be less than %skb.' - % max_size) + raise TweepError(f'File is too big, must be less than {max_size}kb.') f.seek(0) # Reset to beginning of file fp = f @@ -1408,7 +1406,7 @@ class API: if file_type in ['gif', 'jpeg', 'png', 'webp']: file_type = 'image/' + file_type elif file_type not in ['image/gif', 'image/jpeg', 'image/png']: - raise TweepError('Invalid file type for image: %s' % file_type) + raise TweepError(f'Invalid file type for image: {file_type}') if isinstance(filename, str): filename = filename.encode('utf-8') diff --git a/tweepy/auth.py b/tweepy/auth.py index 3cd7efa..b283d65 100644 --- a/tweepy/auth.py +++ b/tweepy/auth.py @@ -67,7 +67,7 @@ class OAuthHandler(AuthHandler): try: url = self._get_oauth_url('request_token') if access_type: - url += '?x_auth_access_type=%s' % access_type + url += f'?x_auth_access_type={access_type}' return self.oauth.fetch_request_token(url) except Exception as e: raise TweepError(e) @@ -172,7 +172,7 @@ class AppAuthHandler(AuthHandler): data = resp.json() if data.get('token_type') != 'bearer': raise TweepError('Expected token_type to equal "bearer", ' - 'but got %s instead' % data.get('token_type')) + f'but got {data.get("token_type")} instead') self._bearer_token = data['access_token'] diff --git a/tweepy/binder.py b/tweepy/binder.py index 757733b..60bdeff 100644 --- a/tweepy/binder.py +++ b/tweepy/binder.py @@ -100,7 +100,7 @@ def bind_api(**config): if arg is None: continue if k in self.session.params: - raise TweepError('Multiple values for parameter %s supplied!' % k) + raise TweepError(f'Multiple values for parameter {k} supplied!') self.session.params[k] = convert_to_utf8_str(arg) @@ -117,7 +117,7 @@ def bind_api(**config): try: value = quote(self.session.params[name]) except KeyError: - raise TweepError('No parameter value found for path variable: %s' % name) + raise TweepError(f'No parameter value found for path variable: {name}') del self.session.params[name] self.path = self.path.replace(variable, value) @@ -132,7 +132,7 @@ def bind_api(**config): # Query the cache if one is available # and this request uses a GET method. if self.use_cache and self.api.cache and self.method == 'GET': - cache_result = self.api.cache.get('%s?%s' % (url, urlencode(self.session.params))) + cache_result = self.api.cache.get(f'{url}?{urlencode(self.session.params)}') # if cache result found and not expired, return it if cache_result: # must restore api reference @@ -158,7 +158,7 @@ def bind_api(**config): sleep_time = self._reset_time - int(time.time()) if sleep_time > 0: if self.wait_on_rate_limit_notify: - log.warning("Rate limit reached. Sleeping for: %d" % sleep_time) + log.warning(f"Rate limit reached. Sleeping for: {sleep_time}") time.sleep(sleep_time + 5) # sleep for few extra sec # if self.wait_on_rate_limit and self._reset_time is not None and \ @@ -166,7 +166,7 @@ def bind_api(**config): # sleep_time = self._reset_time - int(time.time()) # if sleep_time > 0: # if self.wait_on_rate_limit_notify: - # log.warning("Rate limit reached. Sleeping for: %d" % sleep_time) + # log.warning(f"Rate limit reached. Sleeping for: {sleep_time}") # time.sleep(sleep_time + 5) # sleep for few extra sec # Apply authentication @@ -188,7 +188,7 @@ def bind_api(**config): auth=auth, proxies=self.api.proxy) except Exception as e: - raise TweepError('Failed to send request: %s' % e).with_traceback(sys.exc_info()[2]) + raise TweepError(f'Failed to send request: {e}').with_traceback(sys.exc_info()[2]) rem_calls = resp.headers.get('x-rate-limit-remaining') @@ -224,7 +224,7 @@ def bind_api(**config): error_msg, api_error_code = \ self.parser.parse_error(resp.text) except Exception: - error_msg = "Twitter error response: status code = %s" % resp.status_code + error_msg = f"Twitter error response: status code = {resp.status_code}" api_error_code = None if is_rate_limit_error_message(error_msg): @@ -239,7 +239,7 @@ def bind_api(**config): # Store result into cache if one is available. if self.use_cache and self.api.cache and self.method == 'GET' and result: - self.api.cache.store('%s?%s' % (url, urlencode(self.session.params)), result) + self.api.cache.store(f'{url}?{urlencode(self.session.params)}', result) return result diff --git a/tweepy/models.py b/tweepy/models.py index f330a7f..d3bdbcb 100644 --- a/tweepy/models.py +++ b/tweepy/models.py @@ -78,8 +78,8 @@ class Model: return results def __repr__(self): - state = ['%s=%s' % (k, repr(v)) for (k, v) in vars(self).items()] - return '%s(%s)' % (self.__class__.__name__, ', '.join(state)) + state = [f'{k}={v!r}' for (k, v) in vars(self).items()] + return f'{self.__class__.__name__}({", ".join(state)})' class Status(Model): diff --git a/tweepy/parsers.py b/tweepy/parsers.py index cd2fb8b..e8c1abf 100644 --- a/tweepy/parsers.py +++ b/tweepy/parsers.py @@ -47,7 +47,7 @@ class JSONParser(Parser): try: json = json_lib.loads(payload) except Exception as e: - raise TweepError('Failed to parse JSON payload: %s' % e) + raise TweepError(f'Failed to parse JSON payload: {e}') if return_cursors and isinstance(json, dict): if 'next' in json: @@ -87,8 +87,7 @@ class ModelParser(JSONParser): return model = getattr(self.model_factory, method.payload_type) except AttributeError: - raise TweepError('No model for this payload type: ' - '%s' % method.payload_type) + raise TweepError(f'No model for this payload type: {method.payload_type}') json = JSONParser.parse(self, method, payload, return_cursors=return_cursors) if isinstance(json, tuple): diff --git a/tweepy/streaming.py b/tweepy/streaming.py index 8c22fe8..649e70f 100644 --- a/tweepy/streaming.py +++ b/tweepy/streaming.py @@ -246,7 +246,7 @@ class Stream: def _run(self): # Authenticate - url = "https://%s%s" % (self.host, self.url) + url = f"https://{self.host}{self.url}" # Connect and process the stream error_counter = 0 @@ -399,7 +399,7 @@ class Stream: self.session.params = {'delimited': 'length'} if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/user.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/user.json' self.host = 'userstream.twitter.com' if stall_warnings: self.session.params['stall_warnings'] = stall_warnings @@ -411,7 +411,7 @@ class Stream: if len(locations) % 4 != 0: raise TweepError("Wrong number of locations points, " "it has to be a multiple of 4") - self.session.params['locations'] = ','.join(['%.2f' % l for l in locations]) + self.session.params['locations'] = ','.join([f'{l:.2f}' for l in locations]) if track: self.session.params['track'] = ','.join(track).encode(encoding) @@ -421,23 +421,23 @@ class Stream: self.session.params = {'delimited': 'length'} if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/statuses/firehose.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/statuses/firehose.json' if count: - self.url += '&count=%s' % count + self.url += f'&count={count}' self._start(is_async) def retweet(self, is_async=False): self.session.params = {'delimited': 'length'} if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/statuses/retweet.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/statuses/retweet.json' self._start(is_async) def sample(self, is_async=False, languages=None, stall_warnings=False): self.session.params = {'delimited': 'length'} if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/statuses/sample.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/statuses/sample.json' if languages: self.session.params['language'] = ','.join(map(str, languages)) if stall_warnings: @@ -450,7 +450,7 @@ class Stream: self.session.headers['Content-type'] = "application/x-www-form-urlencoded" if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/statuses/filter.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/statuses/filter.json' if follow: self.body['follow'] = ','.join(follow).encode(encoding) if track: @@ -459,7 +459,7 @@ class Stream: if len(locations) % 4 != 0: raise TweepError("Wrong number of locations points, " "it has to be a multiple of 4") - self.body['locations'] = ','.join(['%.4f' % l for l in locations]) + self.body['locations'] = ','.join([f'{l:.4f}' for l in locations]) if stall_warnings: self.body['stall_warnings'] = stall_warnings if languages: @@ -474,7 +474,7 @@ class Stream: self.body = {} if self.running: raise TweepError('Stream object already connected!') - self.url = '/%s/site.json' % STREAM_VERSION + self.url = f'/{STREAM_VERSION}/site.json' self.body['follow'] = ','.join(map(str, follow)) self.body['delimited'] = 'length' if stall_warnings: