GodsDevProject commited on
Commit
2d16b09
·
verified ·
1 Parent(s): fa93750

Create ingest/generic_public_foia.py

Browse files
Files changed (1) hide show
  1. ingest/generic_public_foia.py +25 -10
ingest/generic_public_foia.py CHANGED
@@ -1,14 +1,29 @@
1
- import abc
 
2
  from typing import List, Dict
3
 
4
- class GenericFOIAAdapter(abc.ABC):
5
- source_name = "UNKNOWN"
6
- base_url = ""
7
 
8
- @abc.abstractmethod
9
- async def search(self, query: str) -> List[Dict]:
10
- pass
 
 
 
 
 
 
 
 
 
 
 
11
 
12
- def sync_search(self, query: str) -> List[Dict]:
13
- import asyncio
14
- return asyncio.run(self.search(query))
 
 
 
 
 
 
 
1
+ import time
2
+ import asyncio
3
  from typing import List, Dict
4
 
 
 
 
5
 
6
+ class GenericFOIAAdapter:
7
+ """
8
+ Base class for public-only FOIA adapters.
9
+ Non-abstract to avoid HF instantiation errors.
10
+ """
11
+
12
+ source_name: str = "UNKNOWN"
13
+ base_url: str = ""
14
+ is_stub: bool = True
15
+
16
+ def __init__(self, rate_limit: float = 1.0):
17
+ self.rate_limit = rate_limit
18
+ self._last_call = 0.0
19
+ self.last_latency = 0.0
20
 
21
+ async def _throttle(self):
22
+ delta = time.time() - self._last_call
23
+ if delta < self.rate_limit:
24
+ await asyncio.sleep(self.rate_limit - delta)
25
+ self._last_call = time.time()
26
+
27
+ async def search(self, query: str) -> List[Dict]:
28
+ # Safe default: no results
29
+ return []