-
Notifications
You must be signed in to change notification settings - Fork 0
Builtin Services: HTTP
https://github.com/xellu/nautica-api/tree/main/nautica/services/builtins/http
Nautica3's HTTP API is provided by napi.http. It is built on top of Starlette and Uvicorn, and uses a file-based routing convention inspired by SvelteKit. Route paths are derived from the file's location under src/http/, so you don't need to declare them manually.
from napi.http import HTTP, Context, Reply, Error, Require, ReplyModel, StatusCodesRoutes are defined in .py files under src/http/. The URL path is derived from the file path:
| File | Path |
|---|---|
src/http/users.py |
/users/<function name> |
src/http/api/v1/auth.py |
/api/v1/auth/<function name> |
src/http/api/v1/auth/+root.py |
/api/v1/auth/<function name> |
+root.py acts as the index file for a directory
Use @HTTP.<METHOD>() to mark a function as a route handler. The function name becomes the final path segment, which you can override it by passing a name instead
@HTTP.GET()
async def users(ctx: Context):
return Reply(users=[])
#path: /users@HTTP.GET("all-users")
async def get_users(ctx: Context):
return Reply(users=[])
#path: /all-users@HTTP.GET(...)
@HTTP.POST(...)
@HTTP.PUT(...)
@HTTP.DELETE(...)
@HTTP.PATCH(...)
@HTTP.HEAD(...)
@HTTP.CONNECT(...)
@HTTP.TRACE(...)Nautica3 uses Starlette's path parameters:
@HTTP.GET("/{user_id:str}")
async def get_user(ctx: Context):
user_id = ctx.params["user_id"]
return Reply(id=user_id)See https://starlette.dev/routing/
Every route handler receives a Context object as its first argument (unless the handler takes no arguments at all).
@HTTP.GET()
async def example(ctx: Context): ...| Field | Type | Description |
|---|---|---|
ctx.body |
dict |
JSON request body |
ctx.query |
dict |
URL query parameters |
ctx.headers |
dict |
Request headers |
ctx.cookies |
dict |
Request cookies |
ctx.files |
dict[str, AttachedFile] |
Uploaded files |
ctx.params |
dict |
Path parameters |
ctx.clientIp |
str | None |
Client IP address |
ctx.url |
URL |
Full request URL |
ctx.request |
Request |
Original Starlette request |
Use Reply to return a response from a route handler.
Pass keyword arguments to return a JSON object:
return Reply(ok=True, user="Martin")
# {"ok": true, "user": "Martin"}Pass positional arguments into Reply.list to return a JSON array:
return Reply.list("Martin", "Lucy")
# OR
return Reply("Martin", "Lucy").asList()
#Outputs: ["Martin", "Lucy"]Non-empty list replies will get serialized even without .asList(). But we recommend you to use it anyway, as empty lists will get serialized into a dict instead of a list
Return a tuple of (Reply, status_code):
return Reply(), 201r = Reply(ok=True)
r.SetHeader({"X-Custom-Header": "value"})
return rr = Reply(ok=True)
r.SetCookie("session") \
.value("abc123") \
.maxAge(60 * 60 * 24) \
.httpOnly(True) \
.secure(True) \
.build()
return r.build() writes the cookie onto the Reply and returns it
You can also use the Starlette's reponses:
-
Reply.plainText-> PlainTextResponse -
Reply.html-> HTMLResponse -
Reply.stream-> StreamingResponse -
Reply.file-> FileResponse -
Reply.redirect-> RedirectResponse
But by using this, you'll lose response model validation and access to SetHeader, SetCookie, etc.
Raise Error to return an error response. It can be raised anywhere in the handler, including nested functions
from napi.http import Error, StatusCodes
@HTTP.GET()
async def get_user(ctx: Context):
raise Error(StatusCodes.NOT_FOUND, "User not found")Error(
status_code, # HTTP status code, default 400
errorMessage=None, # Message, falls back to standard HTTP message
details=None # Extra context, dict or str
)raise Error(StatusCodes.CONFLICT, "Username taken", details={"field": "username"})
# Response: {"error": "Username taken", "details": {"field": "username"}}Do not return
Error, raise instead
Use @HTTP.Require() to declare the expected shape of an incoming request. If the request does not match, Nautica3 automatically returns a 422 Unprocessable Content with details about what was missing or invalid, and your handler code is never reached
@HTTP.POST()
@HTTP.Require(
body = {"username": str, "age": int},
query = {"format": Require.AnyOf("json", "csv")},
headers = {"X-Api-Key": str},
cookies = {"session": str}
)
async def example(ctx: Context):
...Note that schemas can be nested
All undefined fields are optional, only declare what your route actually needs.
Instead of a plain type, you can use a Requirement validator as the value:
Value must be one of the provided options:
@HTTP.Require(query = {"format": Require.AnyOf("list", "dict")})Value must match any of the provided types:
@HTTP.Require(body = {"value": Require.AnyTypeOf(str, int)})Value must equal the provided value exactly:
@HTTP.Require(headers = {"X-Version": Require.ExactMatch("v3")})Value must match the provided regex pattern:
@HTTP.Require(body = {"username": Require.RegExMatch(r"^[a-zA-Z0-9_]{3,20}$")})Learn more at: https://github.com/xellu/nautica-api/wiki/Requirement-Validators
Use Require.File in the files field to handle multipart file uploads. Only Require.File can be used in files, and Require.File can only be used in files
@HTTP.POST()
@HTTP.Require(
files = {
"avatar": Require.File(
max_size = Require.File.MB(2),
mime = ["image/png", "image/jpeg"]
)
}
)
async def upload_avatar(ctx: Context):
file = ctx.files["avatar"]
await file.save(f"uploads/{file.filename}")
return Reply(ok=True)| Parameter | Type | Description |
|---|---|---|
max_size |
int |
Maximum file size in bytes |
mime |
list[str] |
Allowed MIME types |
from napi.http.Require import File
File.KB(128) # 128 * 1024 bytes
File.MB(2) # 2 * 1024 * 1024 bytes
File.GB(1) # 1 * 1024 * 1024 * 1024 bytesFiles in ctx.files are AttachedFile instances:
| Field / Method | Description |
|---|---|
file.filename |
Original filename |
file.mime |
MIME type |
file.size |
File size in bytes (updated on read) |
await file.read() |
Returns file contents as bytes
|
await file.save(path) |
Saves file to the given path |
Like how @HTTP.Require checks what's coming in, @HTTP.Responses(...) does the same for outgoing traffic.
@HTTP.GET()
@HTTP.Responses(
ReplyModel(200, {"username": str, "age": int}),
Error(StatusCodes.UNAUTHORIZED) # or 401
)
def get_user(ctx: Context):
...This allows you to define all possible responses with the corresponding status code.
By enabling the strict mode:
@HTTP.GET()
@HTTP.Responses(
...,
strict = True
)
...You'll enforce the output schema, meaning if a malformed response is detected, the request will fail. Otherwise the reply models are used for OpenAPI generation only.
dicts, types and Requirements are all valid shapes for the ReplyModel
Run code before or after a specific route handler using @HTTP.Before and @HTTP.After. These are per-route only.
If the before handler returns a value, it is sent as the response immediately and the main handler is skipped:
def check_auth(ctx: Context):
if not ctx.cookies.get("session"):
return Reply(error="Unauthorized"), 401
@HTTP.GET()
@HTTP.Before(check_auth)
async def protected(ctx: Context):
return Reply(ok=True)After handlers run following the main handler. Return value is ignored:
def log_request(ctx: Context):
print(f"Request from {ctx.clientIp}")
@HTTP.GET()
@HTTP.After(log_request)
async def tracked(ctx: Context):
return Reply(ok=True)StatusCodes exposes all standard HTTP status codes as named constants:
from napi.http import StatusCodes
StatusCodes.OK # 200
StatusCodes.CREATED # 201
StatusCodes.BAD_REQUEST # 400
StatusCodes.UNAUTHORIZED # 401
StatusCodes.NOT_FOUND # 404
StatusCodes.CONFLICT # 409
StatusCodes.INTERNAL_SERVER_ERROR # 500Helper functions are also available:
StatusCodes.isSuccess(200) # True
StatusCodes.isRedirect(301) # True
StatusCodes.isClientError(404) # True
StatusCodes.isServerError(500) # True
StatusCodes.getMessage(404) # "Not Found"# file: src/http/api/v1/users.py
from napi.http import HTTP, Context, Reply, Error, Require, StatusCodes
USERS = {
"martin": {"age": 25, "role": "admin"},
"lucy": {"age": 23, "role": "user"},
}
@HTTP.GET()
@HTTP.Require(
query = {"format": Require.AnyOf("list", "dict")}
)
async def users(ctx: Context):
if ctx.query["format"] == "dict":
return Reply(**USERS)
return Reply(*[{"name": k, **v} for k, v in USERS.items()])
@HTTP.GET("/{username:str}")
@HTTP.Responses(
ReplyModel(200, {"age": int, "role": str}),
Error(StatusCodes.NOT_FOUND)
)
async def get_user(ctx: Context):
user = USERS.get(ctx.params["username"])
if not user:
raise Error(StatusCodes.NOT_FOUND, "User not found")
return Reply(**user)
@HTTP.POST()
@HTTP.Require(
body = {"name": str, "age": int}
)
async def create_user(ctx: Context):
name = ctx.body["name"]
if name in USERS:
raise Error(StatusCodes.CONFLICT, f"User '{name}' already exists")
USERS[name] = {"age": ctx.body["age"], "role": "user"}
return Reply(), StatusCodes.CREATED- Home
- CLI Reference
- Service Registry
- Built-in Services
- Managers
- Requirement Validators