Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
from starlette.applications import Starlette
from starlette.middleware import Middleware
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.responses import PlainTextResponse
from starlette.routing import Route
from starlette.testclient import TestClient
class CustomMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request, call_next):
response = await call_next(request)
response.headers["Custom-Header"] = "Example"
return response
app = Starlette()
app.add_middleware(CustomMiddleware)
@app.route("/")
def homepage(request):
return PlainTextResponse("Homepage")
@app.route("/exc")
def exc(request):
raise Exception()
@app.route("/no-response")
class NoResponse:
def __init__(self, scope, receive, send):
@mock_service.route("/")
def mock_service_endpoint(request):
return JSONResponse({"mock": "example"})
app = Starlette()
@app.route("/")
def homepage(request):
client = TestClient(mock_service)
response = client.get("/")
return JSONResponse(response.json())
startup_error_app = Starlette()
@startup_error_app.on_event("startup")
def startup():
raise RuntimeError()
def test_use_testclient_in_endpoint():
"""
We should be able to use the test client within applications.
This is useful if we need to mock out other services,
during tests or in development.
"""
client = TestClient(app)
response = client.get("/")
from starlette.applications import Starlette
from starlette.testclient import TestClient
from starlette_jsonrpc import dispatcher
from starlette_jsonrpc.endpoint import JSONRPCEndpoint
app = Starlette()
@dispatcher.add_method
async def sum(params):
return {"sum": params["x"] + params["y"]}
@dispatcher.add_method
async def subtract(params):
return params["x"] - params["y"]
@dispatcher.add_method(name="SubtractMethod")
async def second_method(params):
return params["x"] - params["y"]
def test_request_withhold_token(self):
app = App()
app.add_middleware(SessionMiddleware, secret_key="xxxxx")
oauth = OAuth()
client = oauth.register(
"dev",
client_id="dev",
client_secret="dev",
api_base_url="https://i.b/api",
access_token_url="https://i.b/token",
authorize_url="https://i.b/authorize",
)
def fake_send(sess, req, **kwargs):
auth = req.headers.get("Authorization")
self.assertIsNone(auth)
resp = mock.MagicMock()
resp.text = "hi"
def test_cors_allow_specific_origin():
app = Starlette()
app.add_middleware(
CORSMiddleware,
allow_origins=["https://example.org"],
allow_headers=["X-Example", "Content-Type"],
)
@app.route("/")
def homepage(request):
return PlainTextResponse("Homepage", status_code=200)
client = TestClient(app)
# Test pre-flight response
headers = {
"Origin": "https://example.org",
def test_starlette_aws_response(mock_data) -> None:
mock_event = mock_data.get_aws_event()
app = Starlette()
@app.route(mock_event["path"])
def homepage(request):
return PlainTextResponse("Hello, world!")
handler = AWSLambdaAdapter(app)
mock_event["body"] = None
response = handler(mock_event, {})
assert response == {
"statusCode": 200,
"isBase64Encoded": False,
"headers": {
"content-length": "13",
"content-type": "text/plain; charset=utf-8",
},
def __init__(self, program, frontend: str, port: int = 80, prefix="/"):
self.program = program
self.app = Starlette(debug=True)
self.url = get_server_url(program.lifespan, port, prefix)
self.template = TemplateFolder(join(frontend, "templates"))
self.app.add_route("/coffee", self.get_coffee)
self.testclient = WebserverTest(self.app)
self.api = Api(self.app, self.program)
self.make_hooks()
self.app.mount(
"/",
GZipMiddleware(
CacheControlMiddleware(
StaticFiles(directory=join(frontend, "dist"), html=True)
),
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
import uvicorn
from starlette_context import context
from starlette_context.middleware import ContextMiddleware
app = Starlette(debug=True)
app.add_middleware(ContextMiddleware)
@app.route("/")
async def index(request: Request):
# adding some dummy data so it actually has some context
import datetime
import uuid
context.update(
a="b", ts=str(datetime.datetime.utcnow()), uuid=uuid.uuid4().hex
)
return JSONResponse(context.dict())
import json
from urllib.parse import unquote
from collections import OrderedDict
from starlette.applications import Starlette
from starlette.responses import Response
from ..queries import recon_query
from ..db import es
from ..utils import clean_regno
from ..utils import JSONResponseDate as JSONResponse, slugify
from .. import settings
from ..templates import templates
app = Starlette()
@app.route('/propose_properties')
async def propose_properties(request):
properties = {
"properties": [
{
"id": "active",
"name": "Active (True/False)"
},
{
"id": "alternateName",
"name": "List of alternative names"
},
{
"id": "charityNumber",
"name": "Charity Number"
import starlette
from starlette.applications import Starlette
from starlette.responses import JSONResponse, HTMLResponse
import uvicorn
from imagers.avatar_imager import AvatarImager
from imagers.badge_imager import GuildBadgeImager
import os
app = Starlette(debug=True)
@app.route('/')
async def homepage(request):
return JSONResponse({'hello': 'world'})
@app.exception_handler(404)
async def not_found(request, exc):
return HTMLResponse(content="Not Found", status_code=exc.status_code)
@app.exception_handler(500)
async def server_error(request, exc):
return HTMLResponse(content="Uh Oh, thats not supposed to happen...", status_code=exc.status_code)
if __name__ == '__main__':
if not os.path.exists('cache/'):
os.mkdir("cache/")