Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_staticfiles_instantiated_with_missing_directory(tmpdir):
with pytest.raises(RuntimeError) as exc_info:
path = os.path.join(tmpdir, "no_such_directory")
StaticFiles(directory=path)
assert "does not exist" in str(exc_info.value)
def test_staticfiles_prevents_breaking_out_of_directory(tmpdir):
directory = os.path.join(tmpdir, "foo")
os.mkdir(directory)
path = os.path.join(tmpdir, "example.txt")
with open(path, "w") as file:
file.write("outside root dir")
app = StaticFiles(directory=directory)
# We can't test this with 'requests', so we test the app directly here.
path = app.get_path({"path": "/../example.txt"})
scope = {"method": "GET"}
loop = asyncio.get_event_loop()
response = loop.run_until_complete(app.get_response(path, scope))
assert response.status_code == 404
assert response.body == b"Not Found"
def test_staticfiles(tmpdir):
path = os.path.join(tmpdir, "example.txt")
with open(path, "w") as file:
file.write("")
app = StaticFiles(directory=tmpdir)
client = TestClient(app)
response = client.get("/example.txt")
assert response.status_code == 200
assert response.text == ""
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")),
],
)
main_app.mount('/api', app=api)
main_app.mount('/', app=app)
# Bootstrap
if __name__ == '__main__':
console.print('[green]여보세요 ![/]')
uvicorn.run(main_app, host='0.0.0.0', port=8000)
)
# start a task that forwards all data from the terminal to browsers
task = asyncio.ensure_future(
_task_forward_terminal_data_to_web_clients(terminal)
)
done, pending = await (
asyncio.wait([task], return_when=asyncio.FIRST_COMPLETED)
)
for task in pending:
task.cancel()
finally:
terminals.pop(terminal_id, None)
app.mount("/", StaticFiles(directory=PUBLIC_DIR, html=True))
app.mount("/static", StaticFiles(directory=STATIC_DIR, html=True))
from fastai.vision import *
from io import BytesIO
from starlette.applications import Starlette
from starlette.middleware.cors import CORSMiddleware
from starlette.responses import HTMLResponse, JSONResponse
from starlette.staticfiles import StaticFiles
export_file_url = 'https://www.dropbox.com/s/6bgq8t6yextloqp/export.pkl?raw=1'
export_file_name = 'export.pkl'
classes = ['black', 'grizzly', 'teddys']
path = Path(__file__).parent
app = Starlette()
app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_headers=['X-Requested-With', 'Content-Type'])
app.mount('/static', StaticFiles(directory='app/static'))
async def download_file(url, dest):
if dest.exists(): return
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.read()
with open(dest, 'wb') as f:
f.write(data)
async def setup_learner():
await download_file(export_file_url, path / export_file_name)
try:
learn = load_learner(path, export_file_name)
return learn
#! python3
import pathlib
from starlette.applications import Starlette
from starlette.staticfiles import StaticFiles
from starlette.templating import Jinja2Templates
from .config import init_db, global_configs
from .loggers import logger
static_dir = pathlib.Path(__file__).absolute().parent / 'static'
templates_dir = pathlib.Path(__file__).absolute().parent / 'templates'
app = Starlette()
app.mount('/static', StaticFiles(directory=str(static_dir)), name='static')
app.config = global_configs
app.logger = logger
app.db = init_db()
app.templates = Jinja2Templates(directory=str(templates_dir))
@app.on_event('startup')
async def _ensure_article_table_exists():
await app.db._ensure_article_table_exists()
# Note: The entire app is reloaded on JS changes in development
# and in production they are immutable so we can enable the lru
# cache regardless of the environment
@alru_cache(maxsize=None)
async def _hash_file(path: str) -> str:
h = hashlib.sha256()
async with aiofiles.open(path, mode="rb") as file:
while True:
chunk = await file.read(FileResponse.chunk_size)
h.update(chunk)
if len(chunk) < FileResponse.chunk_size:
break
return h.hexdigest()
class StaticFiles(_StaticFiles):
"""
A version of StaticFiles that sets etag based on file contents
"""
def file_response(
self,
full_path: str,
stat_result: os.stat_result,
scope: Scope,
status_code: int = 200,
) -> Response:
method = scope["method"]
return FileResponse(
full_path, status_code=status_code, stat_result=stat_result, method=method
)
adminsite = AdminSite(name="admin", permission_scopes=["authenticated"])
# register admins
adminsite.register(DemoAdmin)
adminsite.register(DemoModelAdmin)
adminsite.register(SystemSettingsModelAdmin)
# register widgets
adminsite.register_widget(Today())
adminsite.register_widget(Time())
adminsite.register_widget(DayOfYear())
# create app
app = Starlette(debug=DEBUG)
app.mount(
path="/static",
app=StaticFiles(directory="static", packages=["starlette_admin"]),
name="static"
)
app.add_middleware(AuthenticationMiddleware, backend=DummyAuthBackend())
app.add_middleware(SessionMiddleware, secret_key="secret")
app.add_middleware(DatabaseMiddleware)
# mount admin site
app.mount(path="/", app=adminsite, name=adminsite.name)
from starlette.responses import HTMLResponse, JSONResponse
from starlette.staticfiles import StaticFiles
from starlette.middleware.cors import CORSMiddleware
import uvicorn, aiohttp, asyncio
from io import BytesIO
from fastai.vision import *
model_file_url = 'https://www.dropbox.com/s/y4kl2gv1akv7y4i/stage-2.pth?raw=1'
model_file_name = 'model'
classes = ['black', 'grizzly', 'teddys']
path = Path(__file__).parent
app = Starlette()
app.add_middleware(CORSMiddleware, allow_origins=['*'], allow_headers=['X-Requested-With', 'Content-Type'])
app.mount('/static', StaticFiles(directory='app/static'))
async def download_file(url, dest):
if dest.exists(): return
async with aiohttp.ClientSession() as session:
async with session.get(url) as response:
data = await response.read()
with open(dest, 'wb') as f: f.write(data)
async def setup_learner():
await download_file(model_file_url, path/'models'/f'{model_file_name}.pth')
data_bunch = ImageDataBunch.single_from_classes(path, classes,
ds_tfms=get_transforms(), size=224).normalize(imagenet_stats)
learn = create_cnn(data_bunch, models.resnet34, pretrained=False)
learn.load(model_file_name)
return learn