22
33from __future__ import annotations
44
5- import random
5+ import secrets
66import time
7- from typing import Any , Literal
7+ from typing import Literal , Protocol , cast
88
99import curl_cffi
1010from pydantic import BaseModel , ConfigDict , Field
1515
1616API_BASE_URL = "https://learning.oreilly.com/api/v2"
1717
18- # ---------------------------------------------------------------------------
19- # Response models
20- # ---------------------------------------------------------------------------
21-
2218
2319class SearchResult (BaseModel ):
2420 model_config = ConfigDict (extra = "allow" )
@@ -78,11 +74,6 @@ class FilesManifest(BaseModel):
7874 results : list [FilesManifestItem ] = Field (default_factory = list )
7975
8076
81- # ---------------------------------------------------------------------------
82- # Helpers
83- # ---------------------------------------------------------------------------
84-
85-
8677def normalize_book_identifier (value : str ) -> str :
8778 cleaned = value .strip ().rstrip ("/" )
8879 if not cleaned :
@@ -96,14 +87,17 @@ def normalize_book_identifier(value: str) -> str:
9687 return cleaned .rsplit ("/" , maxsplit = 1 )[- 1 ]
9788
9889
99- # ---------------------------------------------------------------------------
100- # Client
101- # ---------------------------------------------------------------------------
102-
10390_DEFAULT_DELAY_MIN : float = 0.75
10491_DEFAULT_DELAY_MAX : float = 1.0
10592
10693
94+ class _JsonResponse (Protocol ):
95+ url : str
96+
97+ def raise_for_status (self ) -> None : ...
98+ def json (self ) -> dict [str , str ] | list [dict [str , str ]]: ...
99+
100+
107101class OreillyClient :
108102 """Unauthenticated client for the public O'Reilly Learning API.
109103
@@ -126,97 +120,89 @@ def __init__(
126120 self ._delay_min = delay_min
127121 self ._delay_max = delay_max
128122
129- # ------------------------------------------------------------------
130- # Public API
131- # ------------------------------------------------------------------
132-
133123 def search (self , keyword : str , * , field : SearchField = "title" ) -> SearchResponse :
134- response = curl_cffi .get (
135- f"{ self ._base_url } /search/" ,
136- params = {
137- "formats" : "book" ,
138- "languages" : "en" ,
139- "include_facets" : "false" ,
140- "field" : field ,
141- "query" : keyword ,
142- "sort" : "popularity" ,
143- "order" : "desc" ,
144- "limit" : "10" ,
145- },
146- headers = BROWSER_HEADERS ,
147- http_version = "v2" ,
148- allow_redirects = True ,
149- verify = True ,
150- impersonate = "chrome146" ,
151- timeout = self ._timeout ,
124+ response = cast (
125+ "_JsonResponse" ,
126+ curl_cffi .get (
127+ f"{ self ._base_url } /search/" ,
128+ params = {
129+ "formats" : "book" ,
130+ "languages" : "en" ,
131+ "include_facets" : "false" ,
132+ "field" : field ,
133+ "query" : keyword ,
134+ "sort" : "popularity" ,
135+ "order" : "desc" ,
136+ "limit" : "10" ,
137+ },
138+ headers = BROWSER_HEADERS ,
139+ http_version = "v2" ,
140+ allow_redirects = True ,
141+ verify = True ,
142+ impersonate = "chrome146" ,
143+ timeout = self ._timeout ,
144+ ),
152145 )
153146 response .raise_for_status ()
154- return SearchResponse .model_validate (response .json ())
147+ data = response .json ()
148+ return SearchResponse .model_validate (data )
155149
156150 def fetch_metadata (self , identifier : str ) -> BookMetadata :
157151 book_id = normalize_book_identifier (identifier )
158152 data = self ._get_json (f"{ self ._base_url } /epubs/urn:orm:book:{ book_id } /" )
159153 return BookMetadata .model_validate (data )
160154
161- def fetch_spine (self , metadata : BookMetadata ) -> dict [ str , Any ] :
155+ def fetch_spine (self , metadata : BookMetadata ) -> object :
162156 return self ._get_json (metadata .spine , params = {"limit" : "1000" })
163157
164- def fetch_files (self , metadata : BookMetadata ) -> dict [ str , Any ] :
158+ def fetch_files (self , metadata : BookMetadata ) -> object :
165159 return self ._get_json (metadata .files , params = {"limit" : "10000" })
166160
167161 def fetch_table_of_contents (self , metadata : BookMetadata ) -> object :
168162 return self ._get_json_or_list (metadata .table_of_contents )
169163
170- def fetch_chapters (self , metadata : BookMetadata ) -> dict [ str , Any ] :
164+ def fetch_chapters (self , metadata : BookMetadata ) -> object :
171165 return self ._get_json (metadata .chapters , params = {"limit" : "1000" })
172166
173- # ------------------------------------------------------------------
174- # Internal helpers
175- # ------------------------------------------------------------------
176-
177167 def _random_delay (self ) -> None :
178- time .sleep (random . uniform (self ._delay_min , self ._delay_max )) # noqa: S311
168+ time .sleep (secrets . SystemRandom (). uniform (self ._delay_min , self ._delay_max ))
179169
180170 def _get_json (
181171 self ,
182172 url : str ,
183173 * ,
184174 params : dict [str , str ] | None = None ,
185- ) -> dict [ str , Any ] :
175+ ) -> object :
186176 self ._random_delay ()
187- response = curl_cffi .get (
188- url ,
189- params = params ,
190- headers = BROWSER_HEADERS ,
191- http_version = "v2" ,
192- allow_redirects = True ,
193- verify = True ,
194- impersonate = "chrome146" ,
195- timeout = self ._timeout ,
177+ response = cast (
178+ "_JsonResponse" ,
179+ curl_cffi .get (
180+ url ,
181+ params = params ,
182+ headers = BROWSER_HEADERS ,
183+ http_version = "v2" ,
184+ allow_redirects = True ,
185+ verify = True ,
186+ impersonate = "chrome146" ,
187+ timeout = self ._timeout ,
188+ ),
196189 )
197- print (response .url )
198190 response .raise_for_status ()
199- data = response .json ()
200- if not isinstance (data , dict ):
201- msg = f"Expected object response from { url } "
202- raise TypeError (msg )
203- return data
191+ return response .json ()
204192
205193 def _get_json_or_list (self , url : str ) -> object :
206194 self ._random_delay ()
207- response = curl_cffi .get (
208- url ,
209- headers = BROWSER_HEADERS ,
210- http_version = "v2" ,
211- allow_redirects = True ,
212- verify = True ,
213- impersonate = "chrome146" ,
214- timeout = self ._timeout ,
195+ response = cast (
196+ "_JsonResponse" ,
197+ curl_cffi .get (
198+ url ,
199+ headers = BROWSER_HEADERS ,
200+ http_version = "v2" ,
201+ allow_redirects = True ,
202+ verify = True ,
203+ impersonate = "chrome146" ,
204+ timeout = self ._timeout ,
205+ ),
215206 )
216- print (response .url )
217207 response .raise_for_status ()
218- data = response .json ()
219- if not isinstance (data , (dict , list )):
220- msg = f"Expected object or array response from { url } "
221- raise TypeError (msg )
222- return data
208+ return response .json ()
0 commit comments