-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathclient.py
More file actions
171 lines (139 loc) · 6.32 KB
/
Copy pathclient.py
File metadata and controls
171 lines (139 loc) · 6.32 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
"""
Stack0 Webdata Client (DEPRECATED)
Use `Screenshots` and `Extraction` clients instead.
"""
from __future__ import annotations
import warnings
from typing import TYPE_CHECKING
from ..screenshots.client import AsyncScreenshots, Screenshots
from ..extraction.client import AsyncExtraction, Extraction
from ..screenshots.types import (
CreateScreenshotRequest,
CreateScreenshotResponse,
GetScreenshotRequest,
ListScreenshotsRequest,
ListScreenshotsResponse,
Screenshot,
)
from ..extraction.types import (
CreateExtractionRequest,
CreateExtractionResponse,
ExtractionResult,
GetExtractionRequest,
ListExtractionsRequest,
ListExtractionsResponse,
)
if TYPE_CHECKING:
from .._client.http import AsyncHTTPClient, HTTPClient
class Webdata:
"""
Webdata client for screenshots and extractions.
DEPRECATED: Use `Screenshots` and `Extraction` clients instead.
This class is kept for backward compatibility.
"""
def __init__(self, http: HTTPClient) -> None:
warnings.warn(
"Webdata is deprecated. Use Screenshots and Extraction clients instead.",
DeprecationWarning,
stacklevel=2,
)
self._screenshots = Screenshots(http)
self._extraction = Extraction(http)
# Screenshot methods (delegating to Screenshots client)
def screenshot(self, request: CreateScreenshotRequest) -> CreateScreenshotResponse:
"""Capture a screenshot of a URL"""
return self._screenshots.capture(request)
def get_screenshot(self, request: GetScreenshotRequest) -> Screenshot:
"""Get a screenshot by ID"""
return self._screenshots.get(request)
def list_screenshots(self, request: ListScreenshotsRequest | None = None) -> ListScreenshotsResponse:
"""List screenshots with pagination and filters"""
return self._screenshots.list(request)
def delete_screenshot(self, request: GetScreenshotRequest) -> dict[str, bool]:
"""Delete a screenshot"""
return self._screenshots.delete(request)
def screenshot_and_wait(
self,
request: CreateScreenshotRequest,
poll_interval: float = 1.0,
timeout: float = 60.0,
) -> Screenshot:
"""Capture a screenshot and wait for completion"""
return self._screenshots.capture_and_wait(request, poll_interval=poll_interval, timeout=timeout)
# Extraction methods (delegating to Extraction client)
def extract(self, request: CreateExtractionRequest) -> CreateExtractionResponse:
"""Extract content from a URL"""
return self._extraction.extract(request)
def get_extraction(self, request: GetExtractionRequest) -> ExtractionResult:
"""Get an extraction by ID"""
return self._extraction.get(request)
def list_extractions(self, request: ListExtractionsRequest | None = None) -> ListExtractionsResponse:
"""List extractions with pagination and filters"""
return self._extraction.list(request)
def delete_extraction(self, request: GetExtractionRequest) -> dict[str, bool]:
"""Delete an extraction"""
return self._extraction.delete(request)
def extract_and_wait(
self,
request: CreateExtractionRequest,
poll_interval: float = 1.0,
timeout: float = 60.0,
) -> ExtractionResult:
"""Extract content and wait for completion"""
return self._extraction.extract_and_wait(request, poll_interval=poll_interval, timeout=timeout)
class AsyncWebdata:
"""
Async webdata client for screenshots and extractions.
DEPRECATED: Use `AsyncScreenshots` and `AsyncExtraction` clients instead.
This class is kept for backward compatibility.
"""
def __init__(self, http: AsyncHTTPClient) -> None:
warnings.warn(
"AsyncWebdata is deprecated. Use AsyncScreenshots and AsyncExtraction clients instead.",
DeprecationWarning,
stacklevel=2,
)
self._screenshots = AsyncScreenshots(http)
self._extraction = AsyncExtraction(http)
# Screenshot methods (delegating to AsyncScreenshots client)
async def screenshot(self, request: CreateScreenshotRequest) -> CreateScreenshotResponse:
"""Capture a screenshot of a URL"""
return await self._screenshots.capture(request)
async def get_screenshot(self, request: GetScreenshotRequest) -> Screenshot:
"""Get a screenshot by ID"""
return await self._screenshots.get(request)
async def list_screenshots(self, request: ListScreenshotsRequest | None = None) -> ListScreenshotsResponse:
"""List screenshots with pagination and filters"""
return await self._screenshots.list(request)
async def delete_screenshot(self, request: GetScreenshotRequest) -> dict[str, bool]:
"""Delete a screenshot"""
return await self._screenshots.delete(request)
async def screenshot_and_wait(
self,
request: CreateScreenshotRequest,
poll_interval: float = 1.0,
timeout: float = 60.0,
) -> Screenshot:
"""Capture a screenshot and wait for completion"""
return await self._screenshots.capture_and_wait(request, poll_interval=poll_interval, timeout=timeout)
# Extraction methods (delegating to AsyncExtraction client)
async def extract(self, request: CreateExtractionRequest) -> CreateExtractionResponse:
"""Extract content from a URL"""
return await self._extraction.extract(request)
async def get_extraction(self, request: GetExtractionRequest) -> ExtractionResult:
"""Get an extraction by ID"""
return await self._extraction.get(request)
async def list_extractions(self, request: ListExtractionsRequest | None = None) -> ListExtractionsResponse:
"""List extractions with pagination and filters"""
return await self._extraction.list(request)
async def delete_extraction(self, request: GetExtractionRequest) -> dict[str, bool]:
"""Delete an extraction"""
return await self._extraction.delete(request)
async def extract_and_wait(
self,
request: CreateExtractionRequest,
poll_interval: float = 1.0,
timeout: float = 60.0,
) -> ExtractionResult:
"""Extract content and wait for completion"""
return await self._extraction.extract_and_wait(request, poll_interval=poll_interval, timeout=timeout)