try:
await fetch_url(url)
except Exception as e:
- print("Exception: %s %s" % (e, url))
+ print(f"Exception: {e} {url}")
dead.add(url)
finally:
q.task_done()
base_elems.append(method.upper())
base_elems.append(normalized_url)
base_elems.append(
- "&".join(
- "%s=%s" % (k, _oauth_escape(str(v))) for k, v in sorted(parameters.items())
- )
+ "&".join(f"{k}={_oauth_escape(str(v))}" for k, v in sorted(parameters.items()))
)
base_string = "&".join(_oauth_escape(e) for e in base_elems)
base_elems.append(method.upper())
base_elems.append(normalized_url)
base_elems.append(
- "&".join(
- "%s=%s" % (k, _oauth_escape(str(v))) for k, v in sorted(parameters.items())
- )
+ "&".join(f"{k}={_oauth_escape(str(v))}" for k, v in sorted(parameters.items()))
)
base_string = "&".join(_oauth_escape(e) for e in base_elems)
# have a status bar, such as Safari by default)
params += ' title="%s"' % href
- return '<a href="%s"%s>%s</a>' % (href, params, url)
+ return f'<a href="{href}"{params}>{url}</a>'
# First HTML-escape so that our strings are all safe.
# The regex is modified to avoid character entites other than & so
elif isawaitable(yielded):
return _wrap_awaitable(yielded) # type: ignore
else:
- raise BadYieldError("yielded unknown object %r" % (yielded,))
+ raise BadYieldError(f"yielded unknown object {yielded!r}")
convert_yielded = singledispatch(convert_yielded)
if self.is_client:
assert isinstance(start_line, httputil.RequestStartLine)
self._request_start_line = start_line
- lines.append(utf8("%s %s HTTP/1.1" % (start_line[0], start_line[1])))
+ lines.append(utf8(f"{start_line[0]} {start_line[1]} HTTP/1.1"))
# Client requests with a non-empty body must have either a
# Content-Length or a Transfer-Encoding. If Content-Length is not
# present we'll add our Transfer-Encoding below.
def __repr__(self) -> str:
args = ",".join("%s=%r" % i for i in sorted(self.__dict__.items()))
- return "%s(%s)" % (self.__class__.__name__, args)
+ return f"{self.__class__.__name__}({args})"
class HTTPClientError(Exception):
def __str__(self) -> str:
lines = []
for name, value in self.get_all():
- lines.append("%s: %s\n" % (name, value))
+ lines.append(f"{name}: {value}\n")
return "".join(lines)
__unicode__ = __str__
def __repr__(self) -> str:
attrs = ("protocol", "host", "method", "uri", "version", "remote_ip")
- args = ", ".join(["%s=%r" % (n, getattr(self, n)) for n in attrs])
- return "%s(%s)" % (self.__class__.__name__, args)
+ args = ", ".join([f"{n}={getattr(self, n)!r}" for n in attrs])
+ return f"{self.__class__.__name__}({args})"
class HTTPInputError(Exception):
"""
start = start or 0
end = (end or total) - 1
- return "bytes %s-%s/%s" % (start, end, total)
+ return f"bytes {start}-{end}/{total}"
def _int_or_none(val: str) -> Optional[int]:
out.append(k)
else:
# TODO: quote if necessary.
- out.append("%s=%s" % (k, v))
+ out.append(f"{k}={v}")
return "; ".join(out)
if plural_message is not None:
assert count is not None
msgs_with_ctxt = (
- "%s%s%s" % (context, CONTEXT_SEPARATOR, message),
- "%s%s%s" % (context, CONTEXT_SEPARATOR, plural_message),
+ f"{context}{CONTEXT_SEPARATOR}{message}",
+ f"{context}{CONTEXT_SEPARATOR}{plural_message}",
count,
)
result = self.ngettext(*msgs_with_ctxt)
result = self.ngettext(message, plural_message, count)
return result
else:
- msg_with_ctxt = "%s%s%s" % (context, CONTEXT_SEPARATOR, message)
+ msg_with_ctxt = f"{context}{CONTEXT_SEPARATOR}{message}"
result = self.gettext(msg_with_ctxt)
if CONTEXT_SEPARATOR in result:
# Translation not found
"""
def __repr__(self) -> str:
- result = "<%s" % (self.__class__.__name__,)
+ result = f"<{self.__class__.__name__}"
if self._waiters:
result += " waiters[%s]" % len(self._waiters)
return result + ">"
self._waiters = set() # type: Set[Future[None]]
def __repr__(self) -> str:
- return "<%s %s>" % (
+ return "<{} {}>".format(
self.__class__.__name__,
"set" if self.is_set() else "clear",
)
def __repr__(self) -> str:
res = super().__repr__()
- extra = (
- "locked" if self._value == 0 else "unlocked,value:{0}".format(self._value)
- )
+ extra = "locked" if self._value == 0 else f"unlocked,value:{self._value}"
if self._waiters:
- extra = "{0},waiters:{1}".format(extra, len(self._waiters))
- return "<{0} [{1}]>".format(res[1:-1], extra)
+ extra = f"{extra},waiters:{len(self._waiters)}"
+ return f"<{res[1:-1]} [{extra}]>"
def release(self) -> None:
"""Increment the counter and wake one waiter."""
self._block = BoundedSemaphore(value=1)
def __repr__(self) -> str:
- return "<%s _block=%s>" % (self.__class__.__name__, self._block)
+ return f"<{self.__class__.__name__} _block={self._block}>"
def acquire(
self, timeout: Optional[Union[float, datetime.timedelta]] = None
# byte strings wherever possible).
record.message = _safe_unicode(message)
except Exception as e:
- record.message = "Bad message (%r): %r" % (e, record.__dict__)
+ record.message = f"Bad message ({e!r}): {record.__dict__!r}"
record.asctime = self.formatTime(record, cast(str, self.datefmt))
self._getters.popleft()
def __repr__(self) -> str:
- return "<%s at %s %s>" % (type(self).__name__, hex(id(self)), self._format())
+ return f"<{type(self).__name__} at {hex(id(self))} {self._format()}>"
def __str__(self) -> str:
- return "<%s %s>" % (type(self).__name__, self._format())
+ return f"<{type(self).__name__} {self._format()}>"
def _format(self) -> str:
- result = "maxsize=%r" % (self.maxsize,)
+ result = f"maxsize={self.maxsize!r}"
if getattr(self, "_queue", None):
result += " queue=%r" % self._queue
if self._getters:
return self.matcher.reverse(*args)
def __repr__(self) -> str:
- return "%s(%r, %s, kwargs=%r, name=%r)" % (
+ return "{}({!r}, {}, kwargs={!r}, name={!r})".format(
self.__class__.__name__,
self.matcher,
self.target,
self.kwargs = kwargs
def __repr__(self) -> str:
- return "%s(%r, %s, kwargs=%r, name=%r)" % (
+ return "{}({!r}, {}, kwargs={!r}, name={!r})".format(
self.__class__.__name__,
self.regex.pattern,
self.handler_class,
request, callback, timeout_handle = self.waiting[key]
self.queue.remove((key, request, callback))
- error_message = "Timeout {0}".format(info) if info else "Timeout"
+ error_message = f"Timeout {info}" if info else "Timeout"
timeout_response = HTTPResponse(
request,
599,
if self.request.user_agent:
self.request.headers["User-Agent"] = self.request.user_agent
elif self.request.headers.get("User-Agent") is None:
- self.request.headers["User-Agent"] = "Tornado/{}".format(version)
+ self.request.headers["User-Agent"] = f"Tornado/{version}"
if not self.request.allow_nonstandard_methods:
# Some HTTP methods nearly always have bodies while others
# almost never do. Fail in this case unless the user has
:info string key: More detailed timeout information.
"""
self._timeout = None
- error_message = "Timeout {0}".format(info) if info else "Timeout"
+ error_message = f"Timeout {info}" if info else "Timeout"
if self.final_callback is not None:
self._handle_exception(
HTTPTimeoutError, HTTPTimeoutError(error_message), None
# Reassemble the start line.
self.request.header_callback("%s %s %s\r\n" % first_line)
for k, v in self.headers.get_all():
- self.request.header_callback("%s: %s\r\n" % (k, v))
+ self.request.header_callback(f"{k}: {v}\r\n")
self.request.header_callback("\r\n")
def _should_follow_redirect(self) -> bool:
self.body.generate(writer)
writer.write_line("return _tt_utf8('').join(_tt_buffer)", self.line)
writer.write_line(
- "_tt_append(_tt_utf8(%s(%s())))" % (self.method, method_name), self.line
+ f"_tt_append(_tt_utf8({self.method}({method_name}())))", self.line
)
allowed_parents = intermediate_blocks.get(operator)
if allowed_parents is not None:
if not in_block:
- reader.raise_parse_error(
- "%s outside %s block" % (operator, allowed_parents)
- )
+ reader.raise_parse_error(f"{operator} outside {allowed_parents} block")
if in_block not in allowed_parents:
reader.raise_parse_error(
- "%s block cannot be attached to %s block" % (operator, in_block)
+ f"{operator} block cannot be attached to {in_block} block"
)
body.chunks.append(_IntermediateControlBlock(contents, line))
continue
elif operator in ("break", "continue"):
if not in_loop:
reader.raise_parse_error(
- "%s outside %s block" % (operator, {"for", "while"})
+ "{} outside {} block".format(operator, {"for", "while"})
)
body.chunks.append(_Statement(contents, line))
continue
assert param_dict["nonce"] == nonce
assert param_dict["username"] == self.username
assert param_dict["uri"] == self.request.path
- h1 = md5(
- utf8("%s:%s:%s" % (self.username, realm, self.password))
- ).hexdigest()
- h2 = md5(
- utf8("%s:%s" % (self.request.method, self.request.path))
- ).hexdigest()
- digest = md5(utf8("%s:%s:%s" % (h1, nonce, h2))).hexdigest()
+ h1 = md5(utf8(f"{self.username}:{realm}:{self.password}")).hexdigest()
+ h2 = md5(utf8(f"{self.request.method}:{self.request.path}")).hexdigest()
+ digest = md5(utf8(f"{h1}:{nonce}:{h2}")).hexdigest()
if digest == param_dict["response"]:
self.write("ok")
else:
self.set_status(401)
self.set_header(
"WWW-Authenticate",
- 'Digest realm="%s", nonce="%s", opaque="%s"' % (realm, nonce, opaque),
+ f'Digest realm="{realm}", nonce="{nonce}", opaque="{opaque}"',
)
"WaitIterator dict status incorrect",
)
else:
- self.fail("got bad WaitIterator index {}".format(dg.current_index))
+ self.fail(f"got bad WaitIterator index {dg.current_index}")
i += 1
self.assertIsNone(g.current_index, "bad nil current index")
# Triggering the potential bug seems to depend on input length.
# This length is taken from the bad-response example reported in
# https://github.com/tornadoweb/tornado/pull/2875 (uncompressed).
- text = "".join("Hello World {}\n".format(i) for i in range(9000))[:149051]
+ text = "".join(f"Hello World {i}\n" for i in range(9000))[:149051]
body = gzip.compress(text.encode(), compresslevel=6) + b"\00"
self.write(body)
self.assertLess(abs(response.start_time - start_time), 1.0)
for k, v in response.time_info.items():
- self.assertTrue(0 <= v < 1.0, "time_info[%s] out of bounds: %s" % (k, v))
+ self.assertTrue(0 <= v < 1.0, f"time_info[{k}] out of bounds: {v}")
def test_zero_timeout(self):
response = self.fetch("/hello", connect_timeout=0)
def check_type(self, name, obj, expected_type):
actual_type = type(obj)
if expected_type != actual_type:
- self.errors[name] = "expected %s, got %s" % (expected_type, actual_type)
+ self.errors[name] = f"expected {expected_type}, got {actual_type}"
class PostEchoHandler(RequestHandler):
def test_default_user_agent(self: typing.Any):
response = self.fetch("/user_agent", method="GET")
self.assertEqual(200, response.code)
- self.assertEqual(response.body.decode(), "Tornado/{}".format(version))
+ self.assertEqual(response.body.decode(), f"Tornado/{version}")
def test_see_other_redirect(self: typing.Any):
for code in (302, 303):
)
# tamper with the cookie
handler._cookies["foo"] = utf8(
- "1234|5678%s|%s" % (to_basestring(timestamp), to_basestring(sig))
+ f"1234|5678{to_basestring(timestamp)}|{to_basestring(sig)}"
)
# it gets rejected
with ExpectLog(gen_log, "Cookie timestamp in future"):
def check_type(self, name, obj, expected_type):
actual_type = type(obj)
if expected_type != actual_type:
- self.errors[name] = "expected %s, got %s" % (expected_type, actual_type)
+ self.errors[name] = f"expected {expected_type}, got {actual_type}"
class DecodeArgHandler(RequestHandler):
extension_index = path.rindex(".")
before_version = path[:extension_index]
after_version = path[(extension_index + 1) :]
- return "/static/%s.%s.%s" % (
+ return "/static/{}.{}.{}".format(
before_version,
version_hash,
after_version,
def parse_url_path(self, url_path):
extension_index = url_path.rindex(".")
version_index = url_path.rindex(".", 0, extension_index)
- return "%s%s" % (url_path[:version_index], url_path[extension_index:])
+ return f"{url_path[:version_index]}{url_path[extension_index:]}"
@classmethod
def get_absolute_path(cls, settings, path):
def get_app_kwargs(self):
def my_ui_method(handler, x):
- return "In my_ui_method(%s) with handler value %s." % (x, handler.value())
+ return f"In my_ui_method({x}) with handler value {handler.value()}."
class MyModule(UIModule):
def render(self, x):
- return "In MyModule(%s) with handler value %s." % (
+ return "In MyModule({}) with handler value {}.".format(
x,
typing.cast(UIMethodUIModuleTest.Handler, self.handler).value(),
)
raise NotImplementedError()
def get(self):
- self.write("Hello %s (%s)" % (self.current_user, self.locale.code))
+ self.write(f"Hello {self.current_user} ({self.locale.code})")
def test_set_properties(self):
# Ensure that current_user can be assigned to normally for apps
@contextlib.contextmanager
def in_method(self, method):
if self.method is not None:
- self.test.fail("entered method %s while in %s" % (method, self.method))
+ self.test.fail(f"entered method {method} while in {self.method}")
self.method = method
self.methods.append(method)
try:
def get_url(self, path: str) -> str:
"""Returns an absolute url for the given path on the test server."""
- return "%s://127.0.0.1:%s%s" % (self.get_protocol(), self.get_http_port(), path)
+ return f"{self.get_protocol()}://127.0.0.1:{self.get_http_port()}{path}"
def tearDown(self) -> None:
self.http_server.stop()
return _unicode(value)
except UnicodeDecodeError:
raise HTTPError(
- 400, "Invalid unicode in %s: %r" % (name or "url", value[:40])
+ 400, "Invalid unicode in {}: {!r}".format(name or "url", value[:40])
)
@property
value = escape.native_str(value)
if re.search(r"[\x00-\x20]", name + value):
# Don't let us accidentally inject bad stuff
- raise ValueError("Invalid cookie %r: %r" % (name, value))
+ raise ValueError(f"Invalid cookie {name!r}: {value!r}")
if not hasattr(self, "_new_cookie"):
self._new_cookie = (
http.cookies.SimpleCookie()
self.application.log_request(self)
def _request_summary(self) -> str:
- return "%s %s (%s)" % (
+ return "{} {} ({})".format(
self.request.method,
self.request.uri,
self.request.remote_ip,
# and less than the first-byte-pos.
self.set_status(416) # Range Not Satisfiable
self.set_header("Content-Type", "text/plain")
- self.set_header("Content-Range", "bytes */%s" % (size,))
+ self.set_header("Content-Range", f"bytes */{size}")
return
if end is not None and end > size:
# Clients sometimes blindly use a large range to limit their
version_hash = self._get_cached_version(self.absolute_path)
if not version_hash:
return None
- return '"%s"' % (version_hash,)
+ return f'"{version_hash}"'
def set_headers(self) -> None:
"""Sets the content and caching headers on the response.
if not version_hash:
return url
- return "%s?v=%s" % (url, version_hash)
+ return f"{url}?v={version_hash}"
def parse_url_path(self, url_path: str) -> str:
"""Converts a static URL path into a filesystem path.
projects / thirdparty / tornado.git / commitdiff
