Use `dataclass` for the downloader `Slot` (#7405)

This commit is contained in:
Albert Eduardovich N. 2026-04-08 14:26:01 +03:00 committed by GitHub
parent 830eaeab5d
commit 8835a69f12
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 13 additions and 23 deletions

View File

@ -2,6 +2,7 @@ from __future__ import annotations
import random
from collections import deque
from dataclasses import dataclass, field
from datetime import datetime
from time import time
from typing import TYPE_CHECKING, Any
@ -40,24 +41,21 @@ if TYPE_CHECKING:
from scrapy.signalmanager import SignalManager
@dataclass(slots=True, eq=False)
class Slot:
"""Downloader slot"""
def __init__(
self,
concurrency: int,
delay: float,
randomize_delay: bool,
):
self.concurrency: int = concurrency
self.delay: float = delay
self.randomize_delay: bool = randomize_delay
concurrency: int
delay: float
randomize_delay: bool
self.active: set[Request] = set()
self.queue: deque[tuple[Request, Deferred[Response]]] = deque()
self.transferring: set[Request] = set()
self.lastseen: float = 0
self.latercall: CallLaterResult | None = None
active: set[Request] = field(default_factory=set, init=False, repr=False)
queue: deque[tuple[Request, Deferred[Response]]] = field(
default_factory=deque, init=False, repr=False
)
transferring: set[Request] = field(default_factory=set, init=False, repr=False)
lastseen: float = field(default=0, init=False, repr=False)
latercall: CallLaterResult | None = field(default=None, init=False, repr=False)
def free_transfer_slots(self) -> int:
return self.concurrency - len(self.transferring)
@ -72,14 +70,6 @@ class Slot:
self.latercall.cancel()
self.latercall = None
def __repr__(self) -> str:
cls_name = self.__class__.__name__
return (
f"{cls_name}(concurrency={self.concurrency!r}, "
f"delay={self.delay:.2f}, "
f"randomize_delay={self.randomize_delay!r})"
)
def __str__(self) -> str:
return (
f"<downloader.Slot concurrency={self.concurrency!r} "

View File

@ -35,7 +35,7 @@ if TYPE_CHECKING:
class TestSlot:
def test_repr(self):
slot = Slot(concurrency=8, delay=0.1, randomize_delay=True)
assert repr(slot) == "Slot(concurrency=8, delay=0.10, randomize_delay=True)"
assert repr(slot) == "Slot(concurrency=8, delay=0.1, randomize_delay=True)"
@pytest.mark.requires_reactor # this test is related to the Twisted HTTP code