Endpoint Groups
An endpoint group brings endpoints together: it basically handles routing.
Overview
The endpoint group accepts a list of endpoints/endpoint groups and routes requests to them. You can set a URL for the endpoint group, and this becomes a URL prefix for all of the endpoints under it. Note that all routing is greedy, which means you want to put endpoints with more specific URLs first. Here’s an example of how you can use them to build a fully functional API that manages both users and companies. Each individual endpoint is defined for the purpose of the example, but note that in practice you could accomplish this same thing with much less code by using the RestfulApi endpoint:
import clearskies
from clearskies.validators import Required, Unique
from clearskies import columns
class Company(clearskies.Model):
id_column_name = "id"
backend = clearskies.backends.MemoryBackend()
id = columns.Uuid()
name = columns.String(
validators=[
Required(),
Unique(),
]
)
class User(clearskies.Model):
id_column_name = "id"
backend = clearskies.backends.MemoryBackend()
id = columns.Uuid()
name = columns.String(validators=[Required()])
username = columns.String(
validators=[
Required(),
Unique(),
]
)
age = columns.Integer(validators=[Required()])
created_at = columns.Created()
updated_at = columns.Updated()
company_id = columns.BelongsToId(
Company,
readable_parent_columns=["id", "name"],
validators=[Required()],
)
company = columns.BelongsToModel("company_id")
readable_user_column_names = [
"id",
"name",
"username",
"age",
"created_at",
"updated_at",
"company",
]
writeable_user_column_names = ["name", "username", "age", "company_id"]
users_api = clearskies.EndpointGroup(
[
clearskies.endpoints.Update(
model_class=User,
url="/:id",
readable_column_names=readable_user_column_names,
writeable_column_names=writeable_user_column_names,
),
clearskies.endpoints.Delete(
model_class=User,
url="/:id",
),
clearskies.endpoints.Get(
model_class=User,
url="/:id",
readable_column_names=readable_user_column_names,
),
clearskies.endpoints.Create(
model_class=User,
readable_column_names=readable_user_column_names,
writeable_column_names=writeable_user_column_names,
),
clearskies.endpoints.SimpleSearch(
model_class=User,
readable_column_names=readable_user_column_names,
sortable_column_names=readable_user_column_names,
searchable_column_names=readable_user_column_names,
default_sort_column_name="name",
),
],
url="users",
)
readable_company_column_names = ["id", "name"]
writeable_company_column_names = ["name"]
companies_api = clearskies.EndpointGroup(
[
clearskies.endpoints.Update(
model_class=Company,
url="/:id",
readable_column_names=readable_company_column_names,
writeable_column_names=writeable_company_column_names,
),
clearskies.endpoints.Delete(
model_class=Company,
url="/:id",
),
clearskies.endpoints.Get(
model_class=Company,
url="/:id",
readable_column_names=readable_company_column_names,
),
clearskies.endpoints.Create(
model_class=Company,
readable_column_names=readable_company_column_names,
writeable_column_names=writeable_company_column_names,
),
clearskies.endpoints.SimpleSearch(
model_class=Company,
readable_column_names=readable_company_column_names,
sortable_column_names=readable_company_column_names,
searchable_column_names=readable_company_column_names,
default_sort_column_name="name",
),
],
url="companies",
)
wsgi = clearskies.contexts.WsgiRef(clearskies.EndpointGroup([users_api, companies_api]))
wsgi()
Usage then works exactly as expected:
$ curl 'http://localhost:8080/companies' -d '{"name": "Box Store"}' | jq
{
"status": "success",
"error": "",
"data": {
"id": "f073ee4d-318d-4e0b-a796-f450c40aa771",
"name": "Box Store"
},
"pagination": {},
"input_errors": {}
}
curl 'http://localhost:8080/users' -d '{"name": "Bob Brown", "username": "bobbrown", "age": 25, "company_id": "f073ee4d-318d-4e0b-a796-f450c40aa771"}'
curl 'http://localhost:8080/users' -d '{"name": "Jane Doe", "username": "janedoe", "age": 32, "company_id": "f073ee4d-318d-4e0b-a796-f450c40aa771"}'
$ curl 'http://localhost:8080/users' | jq
{
"status": "success",
"error": "",
"data": [
{
"id": "68cbb9e9-689a-4ae0-af77-d60e4cb344f1",
"name": "Bob Brown",
"username": "bobbrown",
"age": 25,
"created_at": "2025-06-08T10:40:37+00:00",
"updated_at": "2025-06-08T10:40:37+00:00",
"company": {
"id": "f073ee4d-318d-4e0b-a796-f450c40aa771",
"name": "Box Store"
}
},
{
"id": "e69c4ebf-38b1-40d2-b523-5d58f5befc7b",
"name": "Jane Doe",
"username": "janedoe",
"age": 32,
"created_at": "2025-06-08T10:41:04+00:00",
"updated_at": "2025-06-08T10:41:04+00:00",
"company": {
"id": "f073ee4d-318d-4e0b-a796-f450c40aa771",
"name": "Box Store"
}
}
],
"pagination": {
"number_results": 2,
"limit": 50,
"next_page": {}
},
"input_errors": {}
}
endpoints
Required
The list of endpoints connected to this endpoint group
url
Optional
The base URL for the endpoint group.
This URL is added as a prefix to all endpoints attached to the group. This includes any named URL parameters:
response_headers
Optional
Set some response headers that should be returned for this endpoint.
Provide a list of response headers to return to the caller when this endpoint is executed. This should be given a list containing a combination of strings or callables that return a list of strings. The strings in question should be headers formatted as “key: value”. If you attach a callable, it can accept any of the standard dependencies or context-specific values like any other callable in a clearskies application:
def custom_headers(query_parameters):
some_value = "yes" if query_parameters.get("stuff") else "no"
return [f"x-custom: {some_value}", "content-type: application/custom"]
endpoint = clearskies.endpoints.Callable(
lambda: {"hello": "world"},
response_headers=custom_headers,
)
wsgi = clearskies.contexts.WsgiRef(endpoint)
wsgi()
security_headers
Optional
Configure standard security headers to be sent along in the response from this endpoint.
Note that, with CORS, you generally only have to specify the origin. The routing system will automatically add in the appropriate HTTP verbs, and the authorization classes will add in the appropriate headers.
import clearskies
hello_world = clearskies.endpoints.Callable(
lambda: {"hello": "world"},
request_methods=["PATCH", "POST"],
authentication=clearskies.authentication.SecretBearer(environment_key="MY_SECRET"),
security_headers=[
clearskies.security_headers.Hsts(),
clearskies.security_headers.Cors(origin="https://example.com"),
],
)
wsgi = clearskies.contexts.WsgiRef(hello_world)
wsgi()
And then execute the options endpoint to see all the security headers:
$ curl -v http://localhost:8080 -X OPTIONS
* Host localhost:8080 was resolved.
< HTTP/1.0 200 Ok
< Server: WSGIServer/0.2 CPython/3.11.6
< ACCESS-CONTROL-ALLOW-METHODS: PATCH, POST
< ACCESS-CONTROL-ALLOW-HEADERS: Authorization
< ACCESS-CONTROL-MAX-AGE: 5
< ACCESS-CONTROL-ALLOW-ORIGIN: https://example.com
< STRICT-TRANSPORT-SECURITY: max-age=31536000 ;
< CONTENT-TYPE: application/json; charset=UTF-8
< Content-Length: 0
<
* Closing connection
authentication
Optional
The authentication for this endpoint (default is public)
Use this to attach an instance of clearskies.authentication.Authentication
to an endpoint, which enforces authentication. For more details, see the dedicated documentation section on authentication and authorization. By default, all endpoints are public.
authorization
Optional
The authorization rules for this endpoint
Use this to attach an instance of clearskies.authentication.Authorization
to an endpoint, which enforces authorization. For more details, see the dedicated documentation section on authentication and authorization. By default, no authorization is enforced.