silkworm.types

Public type aliases and protocols for annotating silkworm code.

Import from here rather than from private modules:

from silkworm.types import Callback, JSONValue, Logger

async def process_item(self, item: JSONValue, spider: Spider) -> JSONValue: ...

class silkworm.types.BatchItemPipeline

Bases: ItemPipeline, Protocol

Item pipeline that can process an explicit batch in one call.

async process_items(items, spider)

Process a batch in order and return items for the next pipeline.

Parameters:

• items (list[JSONValue])

• spider (Spider)

Return type:

list[JSONValue]

type silkworm.types.BodyData = bytes | bytearray | memoryview | str | Mapping[str, JSONValue] | Iterable[tuple[str, str]] | list[JSONValue] | None

type silkworm.types.Callback = Callable[[Response], Awaitable[None]]

class silkworm.types.CrawlResult

Bases: object

Outcome of a finished crawl, returned by Engine.run() and runners.

spider

Spider name.

Type:

str

close_reason

Why the crawl ended: "finished" when the queue drained, "shutdown" after a stop request or signal, a limit name such as "max_items", or the reason passed to CloseSpider.

Type:

str

elapsed_seconds

Wall-clock crawl duration.

Type:

float

stats

Final counters (see BASE_COUNTERS).

Type:

Mapping[str, int]

labeled_stats

Per-label breakdowns (see BASE_LABELED_COUNTERS).

Type:

Mapping[str, Mapping[str, int]]

custom_stats

A copy of the spider’s stats_payload.

Type:

Mapping[str, JSONValue]

failures

Failure-policy violations; empty when the crawl succeeded.

Type:

tuple[str, …]

property ok: bool

Return whether the crawl met its failure policy.

property requests_sent: int

Return the number of requests handed to the HTTP client.

property responses_received: int

Return the number of responses received.

property items_scraped: int

Return the number of items that passed every pipeline.

property items_dropped: int

Return the number of items discarded by pipelines or limits.

property errors: int

Return the number of unrecovered request or callback failures.

property error_rate: float

Return errors / requests_sent (0.0 when nothing was sent).

property item_drop_rate: float

Return items_dropped / (items_scraped + items_dropped).

__init__(spider, close_reason, elapsed_seconds, stats, labeled_stats, custom_stats=<factory>, failures=())

Parameters:

• spider (str)

• close_reason (str)

• elapsed_seconds (float)

• stats (Mapping[str, int])

• labeled_stats (Mapping[str, Mapping[str, int]])

• custom_stats (Mapping[str, JSONValue])

• failures (tuple[str, ...])

Return type:

None

type silkworm.types.DedupKey = Callable[[Request], str]

class silkworm.types.EngineOptions

Bases: TypedDict

Keyword options for Engine, also accepted by every runner.

run_spider(MySpider, concurrency=32, request_timeout=10) forwards these to Engine; omitted keys use the Engine defaults. Supply http_client to inject a compatible client; its concurrency then controls worker count and default queue capacity.

type silkworm.types.Errback = Callable[[Request, Exception], Awaitable[None]]

class silkworm.types.ExceptionMiddleware

Bases: Protocol

Protocol for middleware that may recover from request failures.

async process_exception(request, exception, spider)

Return a retry request, or None to leave the error unhandled.

Parameters:

• request (Request)

• exception (Exception)

• spider (Spider)

Return type:

Request | None

__init__(*args, **kwargs)

class silkworm.types.FetchClient

Bases: Protocol

Interface the engine needs from an HTTP client.

HttpClient, CDPClient, ServoFetchClient, OnionLinkClient, and CachingHttpClient implement it; pass any conforming object as http_client.

property concurrency: int

Return the maximum number of requests in flight.

property html_max_size_bytes: int

Return the HTML document parsing limit in bytes.

async fetch(req)

Send req and return its response.

Parameters:

req (Request)

Return type:

Response

async close()

Release transport resources.

Return type:

None

__init__(*args, **kwargs)

type silkworm.types.Headers = dict[str, str]

type silkworm.types.ItemCallback = Callable[[JSONValue, Spider], JSONValue | Awaitable[JSONValue | None] | None]

class silkworm.types.ItemPipeline

Bases: Protocol

Protocol implemented by ordered item-processing stages.

async open(spider)

Allocate resources once before the first item is processed.

Parameters:

spider (Spider)

Return type:

None

async close(spider)

Flush buffered data and release resources after the crawl.

Parameters:

spider (Spider)

Return type:

None

async process_item(item, spider)

Process and return the item passed to the next pipeline.

Parameters:

• item (JSONValue)

• spider (Spider)

Return type:

JSONValue

__init__(*args, **kwargs)

type silkworm.types.ItemSchema = ModelSchema | type[object] | ItemValidator

type silkworm.types.ItemValidator = Callable[[JSONValue], JSONValue]

type silkworm.types.JSONLike = JSONScalar | Mapping[str, JSONLike] | Sequence[JSONLike]

type silkworm.types.JSONScalar = str | int | float | bool | None

type silkworm.types.JSONValue = JSONScalar | dict[str, JSONValue] | list[JSONValue]

type silkworm.types.LogLevel = _NormalizedLogLevel | Literal['WARN', 'ERR', 'FATAL'] | None

class silkworm.types.Logger

Bases: Protocol

Structured logger returned by get_logger() and Spider.log.

Keyword arguments become structured context fields on each record.

configure(*, handlers=None)

Replace the logger’s handlers with the supplied configurations.

Each handler mapping accepts sink, level, serialize, and colorize. A sink may be "stderr", "stdout", a filesystem path, or a writable text stream.

Parameters:

handlers (list[dict[str, object]] | None)

Return type:

None

bind(**context)

Return a logger carrying context on every subsequent record.

Parameters:

context (object)

Return type:

Logger

info(message, **context)

Emit an informational record with structured context.

Parameters:

• message (str)

• context (object)

Return type:

None

debug(message, **context)

Emit a diagnostic record with structured context.

Parameters:

• message (str)

• context (object)

Return type:

None

warning(message, **context)

Emit a warning record with structured context.

Parameters:

• message (str)

• context (object)

Return type:

None

error(message, **context)

Emit an error record with structured context.

Parameters:

• message (str)

• context (object)

Return type:

None

exception(message, **context)

Emit an error record including the active exception traceback.

Parameters:

• message (str)

• context (object)

Return type:

None

complete()

Flush all configured handlers.

Return type:

None

__init__(*args, **kwargs)

type silkworm.types.LoopFactory = Callable[[], AbstractEventLoop]

type silkworm.types.MetaData = dict[str, JSONValue]

class silkworm.types.ModelSchema

Bases: Protocol

A Pydantic-style model class: model_validate returns a model instance.

model_validate(obj)

Validate obj and return a model instance, raising on failure.

Parameters:

obj (object)

Return type:

object

__init__(*args, **kwargs)

type silkworm.types.QueryParams = dict[str, QueryValue]

type silkworm.types.QueryValue = str | int | float | bool | None | Iterable[str | int | float | bool | None]

class silkworm.types.RequestMiddleware

Bases: Protocol

Protocol for middleware applied before an HTTP request is sent.

async process_request(request, spider)

Return the request to send, optionally modified or replaced.

Parameters:

• request (Request)

• spider (Spider)

Return type:

Request

__init__(*args, **kwargs)

class silkworm.types.ResponseMiddleware

Bases: Protocol

Protocol for middleware applied after an HTTP response is received.

async process_response(response, spider)

Return a response to dispatch or a request to enqueue instead.

Parameters:

• response (Response)

• spider (Spider)

Return type:

Response | Request

__init__(*args, **kwargs)

type silkworm.types.ZenohKeyResolver = Callable[[JSONValue, Spider], str | Awaitable[str]]