Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 14 additions & 6 deletions src/fastapi_cloud_cli/api/_errors.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,10 +44,18 @@ def _get_response_error_message(response: httpx.Response) -> str | None:
return None # pragma: no cover

detail = data.get("detail")
if not isinstance(detail, str):
return None # pragma: no cover
if isinstance(detail, str):
return detail

if (
isinstance(detail, list)
and detail
and isinstance(detail[0], dict)
and isinstance(message := detail[0].get("msg"), str)
):
return message.removeprefix("Value error, ")

return detail
return None # pragma: no cover


def handle_http_error(
Expand All @@ -61,9 +69,9 @@ def handle_http_error(
if isinstance(error, httpx.HTTPStatusError):
status_code = error.response.status_code

# Handle validation errors from Pydantic models, this should make it easier to debug :)
if status_code == 422:
logger.debug(error.response.json()) # pragma: no cover
logger.debug(error.response.json())
message = _get_response_error_message(error.response)

elif status_code == 400:
message = _get_response_error_message(error.response)
Expand Down Expand Up @@ -103,7 +111,7 @@ def get_http_error_code(error: httpx.HTTPError) -> ErrorCode:
if isinstance(error, httpx.HTTPStatusError):
status_code = error.response.status_code

if status_code in {400, 409}:
if status_code in {400, 409, 422}:
return "invalid_input"

if status_code == 401:
Expand Down
19 changes: 19 additions & 0 deletions src/fastapi_cloud_cli/api/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
AppLogEntry,
BuildLogAdapter,
BuildLogLine,
CustomDomain,
CustomDomainsAPIResponse,
DeploymentStatus,
)
Expand Down Expand Up @@ -141,6 +142,24 @@ def get_custom_domains(self, *, app_id: str) -> CustomDomainsAPIResponse:

return CustomDomainsAPIResponse.model_validate(response.json())

def create_custom_domain(
self,
*,
app_id: str,
name: str,
is_using_pre_validation: bool,
) -> CustomDomain:
response = self.post(
f"/apps/{app_id}/custom-domains",
json={
"name": name,
"is_using_pre_validation": is_using_pre_validation,
},
)
response.raise_for_status()

return CustomDomain.model_validate(response.json())

@attempts(STREAM_LOGS_MAX_RETRIES, STREAM_LOGS_TIMEOUT)
def stream_build_logs(
self, deployment_id: str, *, follow: bool = True
Expand Down
2 changes: 2 additions & 0 deletions src/fastapi_cloud_cli/commands/domains/__init__.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,14 @@
import typer

from fastapi_cloud_cli.commands.domains.add import add_domain
from fastapi_cloud_cli.commands.domains.get import get_domain
from fastapi_cloud_cli.commands.domains.list import list_domains

domains_app = typer.Typer(
no_args_is_help=True,
help="Manage the custom domains of your app.",
)
domains_app.command("add")(add_domain)
domains_app.command("get")(get_domain)
domains_app.command("list")(list_domains)

Expand Down
194 changes: 194 additions & 0 deletions src/fastapi_cloud_cli/commands/domains/add.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,194 @@
from typing import Annotated, Any

import typer
from pydantic import BaseModel, Field
from rich_toolkit import RichToolkit
from rich_toolkit.menu import Option

from fastapi_cloud_cli.api import APIClient, CustomDomain
from fastapi_cloud_cli.commands.domains._shared import _normalize_domain_name
from fastapi_cloud_cli.commands.domains.rendering import render_custom_domain_setup
from fastapi_cloud_cli.utils.apps import resolve_app_id_or_fail
from fastapi_cloud_cli.utils.auth import Identity
from fastapi_cloud_cli.utils.cli import FastAPIRichToolkit, get_rich_toolkit
from fastapi_cloud_cli.utils.execution import JsonOutputOption


class CustomDomainAddOutput(BaseModel):
app_id: str
domain: CustomDomain
show_title: Annotated[bool, Field(exclude=True)] = True


def _resolve_domain_name(
toolkit: FastAPIRichToolkit,
*,
domain: str | None,
) -> str:
if domain is None:
if toolkit.mode == "json":
toolkit.fail(
"missing_required_input",
"Custom domain is required.",
hint="Pass DOMAIN to choose the hostname to add.",
)

domain = toolkit.input(
"What domain do you want to add?",
emoji="🌐",
bullet=False,
)
toolkit.print_line()

return _normalize_domain_name(domain)


def _resolve_pre_validation(
toolkit: FastAPIRichToolkit,
*,
domain: str,
standard: bool,
zero_downtime: bool,
) -> bool:
if standard:
return False
if zero_downtime:
return True
if toolkit.mode == "json":
toolkit.fail(
"missing_required_input",
"Custom domain setup mode is required.",
hint="Pass either --standard or --zero-downtime.",
)

toolkit.print(f"Is {domain} already serving traffic?")
toolkit.print_line()

return toolkit.ask(
"",
options=[
Option(
{
"name": "No — set up a new or unused domain",
"value": False,
}
),
Option(
{
"name": "Yes — migrate it without downtime",
"value": True,
}
),
],
)


def _render_custom_domain_add_output(
data: CustomDomainAddOutput,
toolkit: RichToolkit,
) -> None:
if data.show_title:
toolkit.print_title("custom domains")
toolkit.print_line()

toolkit.print(f"Added [bold]{data.domain.name}[/bold]", emoji="🐔")
render_custom_domain_setup(data.domain, toolkit)


def add_domain(
domain: Annotated[
str | None,
typer.Argument(
help="Hostname of the custom domain to add.",
),
] = None,
app_id: Annotated[
str | None,
typer.Option(
"--app-id",
help="ID of the app to which the custom domain should be added.",
),
] = None,
standard: Annotated[
bool,
typer.Option(
"--standard",
help="Set up a new or unused domain.",
),
] = False,
zero_downtime: Annotated[
bool,
typer.Option(
"--zero-downtime",
help="Migrate an already-live domain without downtime.",
),
] = False,
json_output: JsonOutputOption = False,
) -> Any:
"""
Add a custom domain to an app.
"""
identity = Identity()

with get_rich_toolkit(json_output=json_output) as toolkit:
if not identity.is_logged_in():
toolkit.fail(
"not_logged_in",
"No credentials found.",
hint="Run `fastapi cloud login`.",
)

app_id = resolve_app_id_or_fail(toolkit, app_id=app_id)

if standard and zero_downtime:
toolkit.fail(
"invalid_input",
"Setup modes are mutually exclusive.",
hint="Pass either --standard or --zero-downtime.",
)

prompts_user = domain is None or not (standard or zero_downtime)
if prompts_user and toolkit.mode != "json":
toolkit.print_title("custom domains")
toolkit.print_line()

domain = _resolve_domain_name(toolkit, domain=domain)
is_using_pre_validation = _resolve_pre_validation(
toolkit,
domain=domain,
standard=standard,
zero_downtime=zero_downtime,
)

if prompts_user:
toolkit.print_line()

with APIClient() as client:
with (
toolkit.progress(
title="Adding custom domain",
transient=True,
) as progress,
client.handle_http_errors(
progress,
default_message=(
"Error adding custom domain. Please try again later."
),
not_found_message="App not found.",
toolkit=toolkit,
),
):
created_domain = client.create_custom_domain(
app_id=app_id,
name=domain,
is_using_pre_validation=is_using_pre_validation,
)

toolkit.success(
CustomDomainAddOutput(
app_id=app_id,
domain=created_domain,
show_title=not prompts_user,
),
render_output=_render_custom_domain_add_output,
)
Loading