ddp_utils.ws_intercom.windscribe_selection¶

import ddp_utils.ws_intercom.windscribe_selection

Project-scoped Windscribe location history and random selection policy.

Examples

Use this public operation:

import ddp_utils.ws_intercom.windscribe_selection

Bases: object

Select locations using a project pool, blacklist, and recent history.

History is isolated by project under the DDP script runtime and guarded by both a thread lock and a cross-process FileLock. Plain selectors match country codes, complete fields, region prefixes, ids, and proxy hostnames. Explicit selectors are also supported, for example country:US, region:Canada East, city:Chicago, dc:86, and host:proxy.

Examples

Use this public operation:

instance = WindscribeProjectLocationSelector(...)

Configure one project’s persistent selection policy.

Parameters

Name	Type	Description
project_name	str	Stable project identifier used for runtime isolation.
pool	Optional[WindscribeLocationSelectorInput]	Default allowed selectors. An empty pool allows all locations.
blacklist	Optional[WindscribeLocationSelectorInput]	Selectors that are always removed from the allowed pool.
exclude_recent	int	Number of latest successful locations to exclude.
history_limit	int	Maximum number of successful connection records kept.
history_path	Optional[Union[str, os.PathLike[str]]]	Optional explicit JSON path, primarily for controlled deployments and tests. By default DDP project runtime is used.
lock_timeout	float	Maximum wait for the cross-process history lock.

Raises

Exception	Description
ValueError	If `project_name` is empty, `exclude_recent` is negative, `history_limit` is less than 1 or than `exclude_recent`, or `lock_timeout` is not greater than zero.

property history_path: Path¶

Return the project-specific history JSON path.

Returns

Path of the project-specific history JSON file.

Examples

Use this public operation:

result = instance.history_path()

Apply the pool, blacklist, deduplication, and recent-history window.

Parameters

Name	Type	Description
locations	Sequence[WindscribeLocation]	Candidate locations.
pool	str \| int \| WindscribeLocation \| Iterable[str \| int \| WindscribeLocation] \| None	Allowed selectors; the selector default pool when omitted.
blacklist	str \| int \| WindscribeLocation \| Iterable[str \| int \| WindscribeLocation] \| None	Extra selectors removed on top of the selector blacklist.
exclude_recent	int \| None	Number of latest successful locations to exclude; the selector default when omitted.

Returns

Type	Description
Tuple[WindscribeLocation, …]	Tuple of locations that remain after all filters.

Raises

Exception	Description
ValueError	If `exclude_recent` is negative or exceeds `history_limit`.
WindscribeCatalogError	If no location remains after the pool, blacklist and history filters.

Examples

Use this public operation:

result = instance.eligible_locations(locations=locations_value)

Return one random eligible location without recording it as connected.

Parameters

Name	Type	Description
locations	Sequence[WindscribeLocation]	Candidate locations.
pool	str \| int \| WindscribeLocation \| Iterable[str \| int \| WindscribeLocation] \| None	Allowed selectors; the selector default pool when omitted.
blacklist	str \| int \| WindscribeLocation \| Iterable[str \| int \| WindscribeLocation] \| None	Extra selectors removed on top of the selector blacklist.
exclude_recent	int \| None	Number of latest successful locations to exclude; the selector default when omitted.
rng	Random \| None	Random generator; `random.SystemRandom` when omitted.

Returns

Type	Description
WindscribeLocation	The randomly chosen eligible location.

Examples

Use this public operation:

result = instance.select(locations=locations_value)

record_connection(location: WindscribeLocation, *, selected_host: str | None = None, connected_at: float | None = None) → Dict[str, Any]¶

Persist one successful connection and return its detached history entry.

Parameters

Name	Type	Description
location	WindscribeLocation	Location that was connected successfully.
selected_host	str \| None	Optional host that was selected.
connected_at	float \| None	Unix timestamp of the connection; the current time when omitted.

Returns

Type	Description
Dict[str, Any]	Detached copy of the stored history entry.

Examples

Use this public operation:

result = instance.record_connection(location=location_value)

get_history(*, limit: int | None = None) → List[Dict[str, Any]]¶

Return detached successful-connection records in chronological order.

Parameters

Name	Type	Description
limit	int \| None	Return only this many latest records; all records when omitted.

Returns

Type	Description
List[Dict[str, Any]]	History entries in chronological order.

Raises

Exception	Description
ValueError	If `limit` is negative.

Examples

Use this public operation:

result = instance.get_history()

clear_history() → None¶

Atomically clear only this project’s Windscribe location history.

Examples

Use this public operation:

result = instance.clear_history()