|
1 | 1 | from datetime import timezone |
2 | 2 | from typing import List, TypeAlias, Union |
| 3 | +from urllib.parse import urlencode |
3 | 4 |
|
4 | 5 | from dateutil.parser import parse as dateutil_parse |
5 | 6 | from fastapi import APIRouter, HTTPException, Path, Query, Request |
|
31 | 32 | UserId, |
32 | 33 | ) |
33 | 34 | from birdxplorer_common.storage import Storage |
34 | | -from urllib.parse import urlencode |
35 | 35 |
|
36 | 36 | PostsPaginationMetaWithExamples: TypeAlias = Annotated[ |
37 | 37 | PaginationMeta, |
@@ -279,8 +279,8 @@ def ensure_twitter_timestamp(t: Union[str, TwitterTimestamp]) -> TwitterTimestam |
279 | 279 | try: |
280 | 280 | timestamp = str_to_twitter_timestamp(t) if isinstance(t, str) else t |
281 | 281 | return timestamp |
282 | | - except: |
283 | | - raise ValueError(f"Timestamp out of range") |
| 282 | + except OverflowError: |
| 283 | + raise OverflowError("Timestamp out of range") |
284 | 284 |
|
285 | 285 |
|
286 | 286 | def gen_router(storage: Storage) -> APIRouter: |
@@ -324,7 +324,7 @@ def get_notes( |
324 | 324 | created_at_from = ensure_twitter_timestamp(created_at_from) |
325 | 325 | if created_at_to is not None and isinstance(created_at_to, str): |
326 | 326 | created_at_to = ensure_twitter_timestamp(created_at_to) |
327 | | - except ValueError as e: |
| 327 | + except OverflowError as e: |
328 | 328 | raise HTTPException(status_code=422, detail=str(e)) |
329 | 329 |
|
330 | 330 | notes = list( |
@@ -387,7 +387,7 @@ def get_posts( |
387 | 387 | created_at_from = ensure_twitter_timestamp(created_at_from) |
388 | 388 | if created_at_to is not None and isinstance(created_at_to, str): |
389 | 389 | created_at_to = ensure_twitter_timestamp(created_at_to) |
390 | | - except ValueError as e: |
| 390 | + except OverflowError as e: |
391 | 391 | raise HTTPException(status_code=422, detail=str(e)) |
392 | 392 |
|
393 | 393 | posts = list( |
@@ -467,7 +467,7 @@ def search( |
467 | 467 | note_created_at_from = ensure_twitter_timestamp(note_created_at_from) |
468 | 468 | if note_created_at_to is not None and isinstance(note_created_at_to, str): |
469 | 469 | note_created_at_to = ensure_twitter_timestamp(note_created_at_to) |
470 | | - except ValueError as e: |
| 470 | + except OverflowError as e: |
471 | 471 | raise HTTPException(status_code=422, detail=str(e)) |
472 | 472 |
|
473 | 473 | # Get search results using the optimized storage method |
@@ -533,14 +533,14 @@ def search( |
533 | 533 |
|
534 | 534 | next_url = None |
535 | 535 | if next_offset < total_count: |
536 | | - query_params["offset"] = next_offset |
537 | | - query_params["limit"] = limit |
| 536 | + query_params["offset"] = str(next_offset) |
| 537 | + query_params["limit"] = str(limit) |
538 | 538 | next_url = f"{base_url}?{urlencode(query_params)}" |
539 | 539 |
|
540 | 540 | prev_url = None |
541 | 541 | if offset > 0: |
542 | | - query_params["offset"] = prev_offset |
543 | | - query_params["limit"] = limit |
| 542 | + query_params["offset"] = str(prev_offset) |
| 543 | + query_params["limit"] = str(limit) |
544 | 544 | prev_url = f"{base_url}?{urlencode(query_params)}" |
545 | 545 |
|
546 | 546 | return SearchResponse(data=results, meta=PaginationMeta(next=next_url, prev=prev_url)) |
|
0 commit comments