Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 0 additions & 1 deletion pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -124,7 +124,6 @@ ignore = [
"D213", # Multi-line docstring summary should start at the second line
"D413", # Missing blank line after last section
"EM", # flake8-errmsg
"G004", # Logging statement uses f-string
"ISC001", # This rule may cause conflicts when used with the formatter
"FIX", # flake8-fixme
"PLR0911", # Too many return statements
Expand Down
11 changes: 6 additions & 5 deletions src/apify/_actor.py
Original file line number Diff line number Diff line change
Expand Up @@ -1682,7 +1682,7 @@ async def metamorph(
if not self.configuration.actor_run_id:
raise RuntimeError('actor_run_id cannot be None when running on the Apify platform.')

self.log.debug(f'Metamorphing the Actor run into target Actor {target_actor_id!r}.')
self.log.debug('Metamorphing the Actor run into target Actor %r.', target_actor_id)
await self.apify_client.run(self.configuration.actor_run_id).metamorph(
target_actor_id=target_actor_id,
run_input=run_input,
Expand Down Expand Up @@ -1828,7 +1828,7 @@ async def set_status_message(
"""
if not self.is_at_home():
title = 'Terminal status message' if is_terminal else 'Status message'
self.log.info(f'[{title}]: {status_message}')
self.log.info('[%s]: %s', title, status_message)
return None

# If is_at_home() is True, config.actor_run_id is always set
Expand All @@ -1843,11 +1843,12 @@ async def set_status_message(
return await asyncio.wait_for(update, STATUS_MESSAGE_UPDATE_TIMEOUT.total_seconds())
except TimeoutError:
self.log.warning(
f'Failed to set status message for Actor run with ID "{run_id}": '
f'the update timed out after {STATUS_MESSAGE_UPDATE_TIMEOUT.total_seconds():g}s.'
'Failed to set status message for Actor run with ID "%s": the update timed out after %gs.',
run_id,
STATUS_MESSAGE_UPDATE_TIMEOUT.total_seconds(),
)
except Exception as exc:
self.log.warning(f'Failed to set status message for Actor run with ID "{run_id}": {exc}')
self.log.warning('Failed to set status message for Actor run with ID "%s": %s', run_id, exc)

return None

Expand Down
22 changes: 14 additions & 8 deletions src/apify/_charging.py
Original file line number Diff line number Diff line change
Expand Up @@ -354,12 +354,15 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
# and make the run hit `max_total_charge_usd` early. The key alone decides, as on the platform.
if idempotency_key is not None and (previous := self._idempotent_charges.get(idempotency_key)) is not None:
if previous.event_name == event_name:
logger.debug(f"Skipped a repeated charge of event '{event_name}' under key '{idempotency_key}'.")
logger.debug("Skipped a repeated charge of event '%s' under key '%s'.", event_name, idempotency_key)
repeated_count = previous.charged_count
else:
logger.warning(
f"Idempotency key '{idempotency_key}' was already used to charge for event "
f"'{previous.event_name}', so the charge of event '{event_name}' is skipped."
"Idempotency key '%s' was already used to charge for event '%s', so the charge of event "
"'%s' is skipped.",
idempotency_key,
previous.event_name,
event_name,
)
repeated_count = 0

Expand Down Expand Up @@ -402,13 +405,13 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
count=charged_count,
idempotency_key=idempotency_key,
)
logger.debug(f"Charged {charged_count} occurrence(s) of event '{event_name}'.")
logger.debug("Charged %d occurrence(s) of event '%s'.", charged_count, event_name)
elif event_name in self._tier_priced_events:
logger.warning(
f"Event '{event_name}' is tier-priced and is not chargeable via the pay-per-event API."
"Event '%s' is tier-priced and is not chargeable via the pay-per-event API.", event_name
)
else:
logger.warning(f"Attempting to charge for an unknown event '{event_name}'")
logger.warning("Attempting to charge for an unknown event '%s'", event_name)

# Count the charge only after the API call succeeds, so a failed call retried under the same key is
# counted once.
Expand Down Expand Up @@ -439,8 +442,11 @@ async def charge(self, event_name: str, *, count: int = 1, idempotency_key: str
if charged_count < count:
subject = 'instance' if count == 1 else 'instances'
logger.info(
f"Charging {count} {subject} of '{event_name}' event would exceed max_total_charge_usd "
f'- only {charged_count} events were charged'
"Charging %d %s of '%s' event would exceed max_total_charge_usd - only %d events were charged",
count,
subject,
event_name,
charged_count,
)

return ChargeResult(
Expand Down
10 changes: 5 additions & 5 deletions src/apify/_child_runs.py
Original file line number Diff line number Diff line change
Expand Up @@ -206,10 +206,10 @@ async def find_or_start(
await self._save(name, record.model_copy(update={'abort_with_parent': abort_with_parent}))

if run.status in _RESURRECTABLE_STATUSES:
logger.info(f'Resurrecting child run "{name}"', extra={'run_id': run.id, 'status': run.status})
logger.info('Resurrecting child run "%s"', name, extra={'run_id': run.id, 'status': run.status})
run = await resurrect_run(run.id)
else:
logger.info(f'Reattaching to child run "{name}"', extra={'run_id': run.id, 'status': run.status})
logger.info('Reattaching to child run "%s"', name, extra={'run_id': run.id, 'status': run.status})

await self.update(name, run)
self._clients[name] = client
Expand Down Expand Up @@ -266,12 +266,12 @@ async def _abort(self, name: str, default_client: ApifyClientAsync) -> None:
try:
run = await run_client.abort(gracefully=True)
except NotFoundError:
logger.debug(f'Child run "{name}" no longer exists', extra={'run_id': record.run_id})
logger.debug('Child run "%s" no longer exists', name, extra={'run_id': record.run_id})
except Exception:
logger.exception(f'Failed to abort child run "{name}"', extra={'run_id': record.run_id})
logger.exception('Failed to abort child run "%s"', name, extra={'run_id': record.run_id})
else:
if run.status in _ABORTED_STATUSES:
logger.info(f'Aborted child run "{name}" with the parent', extra={'run_id': record.run_id})
logger.info('Aborted child run "%s" with the parent', name, extra={'run_id': record.run_id})

async def _start(
self,
Expand Down
24 changes: 15 additions & 9 deletions src/apify/events/_apify_event_manager.py
Original file line number Diff line number Diff line change
Expand Up @@ -270,7 +270,9 @@ async def _handle_platform_message(self, message: str | bytes) -> None:

if isinstance(parsed_message, UnknownEvent):
logger.debug(
f'Unknown message received: event_name={parsed_message.name}, event_data={parsed_message.data}'
'Unknown message received: event_name=%s, event_data=%s',
parsed_message.name,
parsed_message.data,
)
return

Expand All @@ -296,21 +298,25 @@ def _should_reconnect_after_close(
"""Log the websocket close and report whether to reconnect (`False` on a non-retryable close code)."""
if websocket.close_code in self._NON_RETRYABLE_CLOSE_CODES:
logger.error(
f'Connection to platform events websocket was closed with a non-retryable code '
f'(code={websocket.close_code}, reason={websocket.close_reason!r}); not reconnecting. '
f'No further platform events will be received for the rest of this run, so migration and '
f'abort handling (e.g. persisting state before a migration) is now disabled.'
'Connection to platform events websocket was closed with a non-retryable code '
'(code=%s, reason=%r); not reconnecting. '
'No further platform events will be received for the rest of this run, so migration and '
'abort handling (e.g. persisting state before a migration) is now disabled.',
websocket.close_code,
websocket.close_reason,
)
return False

if connection_lost:
logger.warning(
f'Connection to platform events websocket was lost '
f'(code={websocket.close_code}, reason={websocket.close_reason!r}), reconnecting...'
'Connection to platform events websocket was lost (code=%s, reason=%r), reconnecting...',
websocket.close_code,
websocket.close_reason,
)
else:
logger.info(
f'Connection to platform events websocket was closed '
f'(code={websocket.close_code}, reason={websocket.close_reason!r}), reconnecting...'
'Connection to platform events websocket was closed (code=%s, reason=%r), reconnecting...',
websocket.close_code,
websocket.close_reason,
)
return True
2 changes: 1 addition & 1 deletion src/apify/scrapy/_async_thread.py
Original file line number Diff line number Diff line change
Expand Up @@ -127,7 +127,7 @@ def wait_for_submitted(self, timeout: timedelta | None = None) -> None:

# Callers rely on the effects having landed, so a timeout has to be visible.
if self._submitted:
logger.warning(f'{len(self._submitted)} submitted coroutines did not finish within the timeout.')
logger.warning('%d submitted coroutines did not finish within the timeout.', len(self._submitted))

def close(self, timeout: timedelta | None = None) -> None:
"""Close the event loop and its thread gracefully.
Expand Down
10 changes: 5 additions & 5 deletions src/apify/scrapy/extensions/_httpcache.py
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,7 @@ async def open_kvs() -> KeyValueStore:

logger.debug("Starting background thread for cache storage's event loop")
self._async_thread = AsyncThread(default_timeout=self._async_thread_timeout)
logger.debug(f"Opening cache storage's {kvs_name!r} key value store")
logger.debug("Opening cache storage's %r key value store", kvs_name)

try:
self._kvs = self._async_thread.run_coro(open_kvs())
Expand Down Expand Up @@ -116,7 +116,7 @@ def close_spider(self, _: Spider, current_time: int | None = None) -> None:
if current_time is None:
current_time = int(time())

logger.info(f'Cleaning up cache items (max {self._expiration_max_items})')
logger.info('Cleaning up cache items (max %s)', self._expiration_max_items)

# Best-effort: a cleanup failure is logged and swallowed (the sweep only reclaims storage, so failing it
# must not turn a normal spider close into an error), and `close` always runs in the `finally`, so
Expand Down Expand Up @@ -190,7 +190,7 @@ def retrieve_response(self, _: Spider, request: Request, current_time: int | Non
headers = Headers(data['headers'])
body = data['body']
except Exception as exc:
logger.warning(f'Ignoring malformed cache entry {key!r}: {exc}', extra={'request': request})
logger.warning('Ignoring malformed cache entry %r: %s', key, exc, extra={'request': request})
return None

respcls = responsetypes.from_args(headers=headers, url=url, body=body)
Expand Down Expand Up @@ -250,11 +250,11 @@ async def _expire_kvs(self, current_time: int) -> None:
try:
gzip_time = read_gzip_time(value)
except Exception as exc:
logger.warning(f'Malformed cache item {item.key}: {exc}')
logger.warning('Malformed cache item %s: %s', item.key, exc)
await self._kvs.delete_value(item.key)
else:
if self._expiration_secs < current_time - gzip_time:
logger.debug(f'Expired cache item {item.key}')
logger.debug('Expired cache item %s', item.key)
await self._kvs.delete_value(item.key)


Expand Down
5 changes: 3 additions & 2 deletions src/apify/scrapy/middlewares/apify_proxy.py
Original file line number Diff line number Diff line change
Expand Up @@ -99,8 +99,9 @@ def process_exception(
"""
if isinstance(exception, TunnelError):
Actor.log.warning(
f'ApifyHttpProxyMiddleware: TunnelError occurred for request="{request}", '
f'reason="{exception}", skipping...'
'ApifyHttpProxyMiddleware: TunnelError occurred for request="%s", reason="%s", skipping...',
request,
exception,
)

async def _get_new_proxy_url(self) -> ParseResult:
Expand Down
9 changes: 5 additions & 4 deletions src/apify/scrapy/requests.py
Original file line number Diff line number Diff line change
Expand Up @@ -165,13 +165,13 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
)
else:
logger.warning(
f'Invalid scrapy_request.headers type, not scrapy.http.headers.Headers: {scrapy_request.headers}'
'Invalid scrapy_request.headers type, not scrapy.http.headers.Headers: %s', scrapy_request.headers
)

apify_request = ApifyRequest.from_url(**request_kwargs)

except Exception:
logger.exception(f'Conversion of Scrapy request {scrapy_request} to Apify request failed; skipping it.')
logger.exception('Conversion of Scrapy request %s to Apify request failed; skipping it.', scrapy_request)
return None

# Serialize the Scrapy request as JSON under 'scrapy_request'. Kept outside the broad except above so
Expand All @@ -181,8 +181,9 @@ def to_apify_request(scrapy_request: ScrapyRequest, spider: Spider) -> ApifyRequ
scrapy_request_json = encode_to_json(scrapy_request_dict)
except (TypeError, ValueError):
logger.exception(
f'Failed to serialize Scrapy request {scrapy_request} for storage on the Apify platform; skipping it. '
'Ensure all values in `meta` and `cb_kwargs` are JSON-serializable.'
'Failed to serialize Scrapy request %s for storage on the Apify platform; skipping it. '
'Ensure all values in `meta` and `cb_kwargs` are JSON-serializable.',
scrapy_request,
)
return None

Expand Down
Loading
Loading