File size: 2,861 Bytes
21b7374 | 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 | import contextlib
import logging
from typing import Any, Dict, Generator, List, Optional, Tuple
import typer
from httpx import HTTPError, HTTPStatusError, ReadTimeout
from rich.segment import Segment
from rich_toolkit import RichToolkit, RichToolkitTheme
from rich_toolkit.progress import Progress
from rich_toolkit.styles import MinimalStyle, TaggedStyle
logger = logging.getLogger(__name__)
class FastAPIStyle(TaggedStyle):
def __init__(self, tag_width: int = 11):
super().__init__(tag_width=tag_width)
def _get_tag_segments(
self,
metadata: Dict[str, Any],
is_animated: bool = False,
done: bool = False,
) -> Tuple[List[Segment], int]:
if not is_animated:
return super()._get_tag_segments(metadata, is_animated, done)
emojis = [
"🥚",
"🐣",
"🐤",
"🐥",
"🐓",
"🐔",
]
tag = emojis[self.animation_counter % len(emojis)]
if done:
tag = emojis[-1]
left_padding = self.tag_width - 1
left_padding = max(0, left_padding)
return [Segment(tag)], left_padding
def get_rich_toolkit(minimal: bool = False) -> RichToolkit:
style = MinimalStyle() if minimal else FastAPIStyle(tag_width=11)
theme = RichToolkitTheme(
style=style,
theme={
"tag.title": "white on #009485",
"tag": "white on #007166",
"placeholder": "grey85",
"text": "white",
"selected": "#007166",
"result": "grey85",
"progress": "on #007166",
"error": "red",
},
)
return RichToolkit(theme=theme)
@contextlib.contextmanager
def handle_http_errors(
progress: Progress,
message: Optional[str] = None,
) -> Generator[None, None, None]:
try:
yield
except ReadTimeout as e:
logger.debug(e)
progress.set_error(
"The request to the FastAPI Cloud server timed out. Please try again later."
)
raise typer.Exit(1) from None
except HTTPError as e:
logger.debug(e)
# Handle validation errors from Pydantic models, this should make it easier to debug :)
if isinstance(e, HTTPStatusError) and e.response.status_code == 422:
logger.debug(e.response.json()) # pragma: no cover
if isinstance(e, HTTPStatusError) and e.response.status_code in (401, 403):
message = "The specified token is not valid. Use `fastapi login` to generate a new token."
else:
message = (
message
or f"Something went wrong while contacting the FastAPI Cloud server. Please try again later. \n\n{e}"
)
progress.set_error(message)
raise typer.Exit(1) from None
|