Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
return Response(content, media_type="text/plain")
app = Router(
[
Route("/", endpoint=homepage, methods=["GET"]),
Mount(
"/users",
routes=[
Route("/", endpoint=users),
Route("/me", endpoint=user_me),
Route("/{username}", endpoint=user),
Route("/nomatch", endpoint=user_no_match),
],
),
Mount("/static", app=Response("xxxxx", media_type="image/png")),
]
)
@app.route("/func")
def func_homepage(request):
return Response("Hello, world!", media_type="text/plain")
@app.route("/func", methods=["POST"])
def contact(request):
return Response("Hello, POST!", media_type="text/plain")
@app.route("/int/{param:int}", name="int-convertor")
def int_convertor(request):
def user_me(request):
content = "User fixed me"
return Response(content, media_type="text/plain")
def user_no_match(request): # pragma: no cover
content = "User fixed no match"
return Response(content, media_type="text/plain")
app = Router(
[
Route("/", endpoint=homepage, methods=["GET"]),
Mount(
"/users",
routes=[
Route("/", endpoint=users),
Route("/me", endpoint=user_me),
Route("/{username}", endpoint=user),
Route("/nomatch", endpoint=user_no_match),
],
),
Mount("/static", app=Response("xxxxx", media_type="image/png")),
]
)
@app.route("/func")
def func_homepage(request):
return Response("Hello, world!", media_type="text/plain")
== "https://foo.example.org/homepage"
)
async def echo_urls(request):
return JSONResponse(
{
"index": request.url_for("index"),
"submount": request.url_for("mount:submount"),
}
)
echo_url_routes = [
Route("/", echo_urls, name="index", methods=["GET"]),
Mount(
"/submount",
name="mount",
routes=[Route("/", echo_urls, name="submount", methods=["GET"])],
),
]
def test_url_for_with_root_path():
app = Starlette(routes=echo_url_routes)
client = TestClient(app, base_url="https://www.example.org/", root_path="/sub_path")
response = client.get("/")
assert response.json() == {
"index": "https://www.example.org/sub_path/",
"submount": "https://www.example.org/sub_path/submount/",
}
response = client.get("/submount/")
while True:
async for frame in camera.frames():
data = b"".join(
[
b"--frame\r\n",
b"Content-Type: image/jpeg\r\n\r\n",
frame,
b"\r\n",
]
)
await send(
{"type": "http.response.body", "body": data, "more_body": True}
)
routes = [Route("/", endpoint=homepage), Mount("/stream/", stream)]
app = Starlette(debug=True, routes=routes)
Route("/500", endpoints.error),
Route("/{username}", endpoints.profile, name="profile", methods=["GET", "POST"]),
Route("/{username}/tables/{table_id}", endpoints.table, name="table", methods=["GET", "POST"]),
Route("/{username}/tables/{table_id}/columns", endpoints.columns, name="columns", methods=["GET", "POST"]),
Route("/{username}/tables/{table_id}/delete", endpoints.delete_table, name="delete-table", methods=["POST"]),
Route("/{username}/tables/{table_id}/upload", endpoints.upload, name="upload", methods=["POST"]),
Route("/{username}/tables/{table_id}/columns/{column_id}/delete", endpoints.delete_column, name="delete-column", methods=["POST"]),
Route("/{username}/tables/{table_id}/{row_uuid}", endpoints.detail, name="detail", methods=["GET", "POST"]),
Route("/{username}/tables/{table_id}/{row_uuid}/delete", endpoints.delete_row, name="delete-row", methods=["POST"]),
Mount("/static", statics, name="static"),
Mount("/auth", routes=auth_routes, name='auth'),
]
if settings.MOCK_GITHUB:
routes += [
Mount("/mock_github", routes=github_routes, name='mock_github')
]
github_client = httpx.AsyncClient(
base_url='http://mock',
app=Router(routes=github_routes)
)
github_api_client = httpx.AsyncClient(
base_url='http://mock',
app=Router(routes=github_routes)
)
GITHUB_AUTH_URL = '/mock_github/login/oauth/authorize'
else: # pragma: nocover
github_client = httpx.AsyncClient(base_url='https://github.com/')
github_api_client = httpx.AsyncClient(
base_url='https://api.github.com/',
headers={'accept': 'application/vnd.github.v3+json'}
'res': res,
'term': request.query_params.get("q"),
'pages': pagination(p["p"], p["size"], res.get("total")),
'selected_org_type': orgtype.get('slug'),
})
routes = [
Route('/', index),
Route('/about', about),
Route('/random', randcharity.random, name='random_org'),
Route('/random.{filetype}', randcharity.random),
Route('/reconcile', reconcile.reconcile_index, methods=['GET', 'POST']),
Mount('/static', StaticFiles(directory="static"), name='static'),
Mount('/reconcile', reconcile.app),
Mount('/charity', charity.app),
Mount('/autocomplete', autocomplete.app),
Mount('/feeds', feeds.app),
Mount('/orgid', routes=orgid.routes),
Mount('/adddata', csvdata.app),
Mount('/admin', routes=admin.routes),
]
middleware = [
Middleware(
CORSMiddleware,
allow_origins=['*'],
allow_methods=['GET', 'POST', 'OPTIONS'],
allow_headers=['Origin', 'Accept', 'Content-Type', 'X-Requested-With', 'X-CSRF-Token'],
)
]
app = Starlette(routes=routes, debug=settings.DEBUG, middleware=middleware)
'term': request.query_params.get("q"),
'pages': pagination(p["p"], p["size"], res.get("total")),
'selected_org_type': orgtype.get('slug'),
})
routes = [
Route('/', index),
Route('/about', about),
Route('/random', randcharity.random, name='random_org'),
Route('/random.{filetype}', randcharity.random),
Route('/reconcile', reconcile.reconcile_index, methods=['GET', 'POST']),
Mount('/static', StaticFiles(directory="static"), name='static'),
Mount('/reconcile', reconcile.app),
Mount('/charity', charity.app),
Mount('/autocomplete', autocomplete.app),
Mount('/feeds', feeds.app),
Mount('/orgid', routes=orgid.routes),
Mount('/adddata', csvdata.app),
Mount('/admin', routes=admin.routes),
]
middleware = [
Middleware(
CORSMiddleware,
allow_origins=['*'],
allow_methods=['GET', 'POST', 'OPTIONS'],
allow_headers=['Origin', 'Accept', 'Content-Type', 'X-Requested-With', 'X-CSRF-Token'],
)
]
app = Starlette(routes=routes, debug=settings.DEBUG, middleware=middleware)
if not hasattr(route.endpoint, method):
continue
func = getattr(route.endpoint, method)
endpoints_info[path].append(
EndpointInfo(
path=path,
method=method.lower(),
func=func,
query_fields=route.query_fields.get(method.upper(), {}),
path_fields=route.path_fields.get(method.upper(), {}),
body_field=route.body_field.get(method.upper()),
output_field=route.output_field.get(method.upper()),
)
)
elif isinstance(route, routing.Mount):
endpoints_info.update(self.get_endpoints(route.routes, base_path=route.path))
return endpoints_info
def mount(self, path: str, app: ASGIApp, name: str = None) -> None:
if isinstance(app, Router):
app.components = self.components
path = path.rstrip("/")
route = Mount(path, app=app, name=name)
self.routes.append(route)
return templates.TemplateResponse('search.html', {
'request': request,
'res': res,
'term': request.query_params.get("q"),
'pages': pagination(p["p"], p["size"], res.get("total")),
'selected_org_type': orgtype.get('slug'),
})
routes = [
Route('/', index),
Route('/about', about),
Route('/random', randcharity.random, name='random_org'),
Route('/random.{filetype}', randcharity.random),
Route('/reconcile', reconcile.reconcile_index, methods=['GET', 'POST']),
Mount('/static', StaticFiles(directory="static"), name='static'),
Mount('/reconcile', reconcile.app),
Mount('/charity', charity.app),
Mount('/autocomplete', autocomplete.app),
Mount('/feeds', feeds.app),
Mount('/orgid', routes=orgid.routes),
Mount('/adddata', csvdata.app),
Mount('/admin', routes=admin.routes),
]
middleware = [
Middleware(
CORSMiddleware,
allow_origins=['*'],
allow_methods=['GET', 'POST', 'OPTIONS'],
allow_headers=['Origin', 'Accept', 'Content-Type', 'X-Requested-With', 'X-CSRF-Token'],
)
]