Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
assert (
mixed_hosts_app.url_path_for("users").make_absolute_url("https://whatever")
== "https://www.example.org/users"
)
assert (
mixed_hosts_app.url_path_for("api:users").make_absolute_url("https://whatever")
== "https://api.example.org/users"
)
async def subdomain_app(scope, receive, send):
response = JSONResponse({"subdomain": scope["path_params"]["subdomain"]})
await response(scope, receive, send)
subdomain_app = Router(
routes=[Host("{subdomain}.example.org", app=subdomain_app, name="subdomains")]
)
def test_subdomain_routing():
client = TestClient(subdomain_app, base_url="https://foo.example.org/")
response = client.get("/")
assert response.status_code == 200
assert response.json() == {"subdomain": "foo"}
def test_subdomain_reverse_urls():
assert (
subdomain_app.url_path_for(
"subdomains", subdomain="foo", path="/homepage"
def test_reverse_mount_urls():
mounted = Router([Mount("/users", ok, name="users")])
assert mounted.url_path_for("users", path="/a") == "/users/a"
users = Router([Route("/{username}", ok, name="user")])
mounted = Router([Mount("/{subpath}/users", users, name="users")])
assert (
mounted.url_path_for("users:user", subpath="test", username="tom")
== "/test/users/tom"
)
assert (
mounted.url_path_for("users", subpath="test", path="/tom") == "/test/users/tom"
)
async def test_lifespan_manager(
startup_exception: typing.Optional[typing.Type[BaseException]],
body_exception: typing.Optional[typing.Type[BaseException]],
shutdown_exception: typing.Optional[typing.Type[BaseException]],
) -> None:
router = Router()
# Setup failing event handlers.
if startup_exception is not None:
@router.on_event("startup")
async def startup() -> None:
assert startup_exception is not None # Please mypy.
raise startup_exception
if shutdown_exception is not None:
@router.on_event("shutdown")
async def shutdown() -> None:
assert shutdown_exception is not None # Please mypy.
raise shutdown_exception
def http_endpoint(request):
url = request.url_for("http_endpoint")
return Response(f"URL: {url}", media_type="text/plain")
class WebSocketEndpoint:
async def __call__(self, scope, receive, send):
websocket = WebSocket(scope=scope, receive=receive, send=send)
await websocket.accept()
await websocket.send_json({"URL": str(websocket.url_for("websocket_endpoint"))})
await websocket.close()
mixed_protocol_app = Router(
routes=[
Route("/", endpoint=http_endpoint),
WebSocketRoute("/", endpoint=WebSocketEndpoint(), name="websocket_endpoint"),
]
)
def test_protocol_switch():
client = TestClient(mixed_protocol_app)
response = client.get("/")
assert response.status_code == 200
assert response.text == "URL: http://testserver/"
with client.websocket_connect("/") as session:
assert session.receive_json() == {"URL": "ws://testserver/"}
def test_raise_on_startup():
def run_startup():
raise RuntimeError()
router = Router(on_startup=[run_startup])
async def app(scope, receive, send):
async def _send(message):
nonlocal startup_failed
if message["type"] == "lifespan.startup.failed":
startup_failed = True
return await send(message)
await router(scope, receive, _send)
startup_failed = False
with pytest.raises(RuntimeError):
with TestClient(app):
pass # pragma: nocover
assert startup_failed
from starlette.endpoints import HTTPEndpoint, WebSocketEndpoint
from starlette.responses import PlainTextResponse
from starlette.routing import Route, Router
from starlette.testclient import TestClient
class Homepage(HTTPEndpoint):
async def get(self, request):
username = request.path_params.get("username")
if username is None:
return PlainTextResponse("Hello, world!")
return PlainTextResponse(f"Hello, {username}!")
app = Router(
routes=[Route("/", endpoint=Homepage), Route("/{username}", endpoint=Homepage)]
)
client = TestClient(app)
def test_http_endpoint_route():
response = client.get("/")
assert response.status_code == 200
assert response.text == "Hello, world!"
def test_http_endpoint_route_path_params():
response = client.get("/tomchristie")
assert response.status_code == 200
assert response.text == "Hello, tomchristie!"
trace = f"switch mastodon trigger {trigger_id}"
elif 'switch_type' in request.path_params and \
request.path_params['switch_type'] == 'mail':
await trigger.update(mail=not trigger.mail)
trace = f"switch mail trigger {trigger_id}"
content = {'errors': ''}
console.print(trace, style="blue")
else:
content = {'errors': {'message': 'Trigger id is missing'}}
console.print(f"error during switch status trigger", style="red")
return JSONResponse(content)
# The API Routes
api = Router(routes=[
Mount('/yeoboseyo', app=Router([
Route('/', endpoint=get_all, methods=['GET']),
Route('/{trigger_id}', endpoint=get, methods=['GET']),
Route('/', endpoint=create, methods=['POST']),
Route('/{trigger_id}', endpoint=update, methods=['PATCH']),
Route('/{trigger_id}', endpoint=delete, methods=['DELETE']),
Route('/switch/{switch_type}/{trigger_id:int}', switch, methods=['PATCH'], name='switch'),
]))
])
app = Starlette(
debug=True,
routes=[
Route('/', homepage, methods=['GET'], name='homepage'),
Mount('/static', StaticFiles(directory="static")),
],
def serverStart():
from starlette.applications import Starlette
from starlette.routing import Router
from starlette.routing import Route
import uvicorn
app = Router([Route('/', endpoint=graphQL, methods=['GET', 'POST'])])
uvicorn.run(app, host='0.0.0.0', port=8000)
return None
"exc": None,
"app": app,
"path_params": route_scope["path_params"],
"route": route,
"websocket": websockets.WebSocket(scope, receive, send),
}
injected_func = await app.injector.inject(endpoint, state)
kwargs = scope.get("kwargs", {})
await injected_func(**kwargs)
return _app
class Router(starlette.routing.Router):
def __init__(self, components: typing.Optional[typing.List[Component]] = None, *args, **kwargs):
super().__init__(*args, **kwargs)
if components is None:
components = []
self.components = components
def add_route(
self,
path: str,
endpoint: typing.Callable,
methods: typing.List[str] = None,
name: str = None,
include_in_schema: bool = True,
):
self.routes.append(
def get_route_handler(self) -> Callable:
return get_request_handler(
dependant=self.dependant,
body_field=self.body_field,
status_code=self.status_code,
response_class=self.response_class or JSONResponse,
response_field=self.secure_cloned_response_field,
response_model_include=self.response_model_include,
response_model_exclude=self.response_model_exclude,
response_model_by_alias=self.response_model_by_alias,
response_model_exclude_unset=self.response_model_exclude_unset,
dependency_overrides_provider=self.dependency_overrides_provider,
)
class APIRouter(routing.Router):
def __init__(
self,
routes: List[routing.BaseRoute] = None,
redirect_slashes: bool = True,
default: ASGIApp = None,
dependency_overrides_provider: Any = None,
route_class: Type[APIRoute] = APIRoute,
default_response_class: Type[Response] = None,
) -> None:
super().__init__(
routes=routes, redirect_slashes=redirect_slashes, default=default
)
self.dependency_overrides_provider = dependency_overrides_provider
self.route_class = route_class
self.default_response_class = default_response_class