Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def file_route(request, filename):
file_path = os.path.join(static_file_directory, filename)
file_path = os.path.abspath(unquote(file_path))
stats = await async_os.stat(file_path)
headers = dict()
headers["Accept-Ranges"] = "bytes"
headers["Content-Length"] = str(stats.st_size)
if request.method == "HEAD":
return HTTPResponse(
headers=headers,
content_type=guess_type(file_path)[0] or "text/plain",
)
else:
return file(
file_path,
headers=headers,
mime_type=guess_type(file_path)[0] or "text/plain",
)
async def file_route(request, filename):
file_path = os.path.join(static_file_directory, filename)
file_path = os.path.abspath(unquote(file_path))
headers = dict()
headers["Accept-Ranges"] = "bytes"
if request.method == "HEAD":
# Return a normal HTTPResponse, not a
# StreamingHTTPResponse for a HEAD request
stats = await async_os.stat(file_path)
headers["Content-Length"] = str(stats.st_size)
return HTTPResponse(
headers=headers,
content_type=guess_type(file_path)[0] or "text/plain",
)
else:
return file_stream(
file_path,
chunk_size=32,
headers=headers,
mime_type=guess_type(file_path)[0] or "text/plain",
)
def test_stat():
"""Test the stat call."""
filename = join(dirname(__file__), 'resources', 'test_file1.txt')
stat_res = yield from aiofiles.os.stat(filename)
assert stat_res.st_size == 10
async def _get_file_size(full_file_path):
stat = await async_os.stat(full_file_path)
return stat.st_size
import aiohttp
import attr
from aiohttp import ClientRequest, hdrs, helpers
from async_exit_stack import AsyncExitStack
from async_generator import async_generator, asynccontextmanager, yield_
from async_lru import alru_cache
from oauthlib.oauth1 import Client as OAuth1Client
from pyrsistent import freeze
from yarl import URL
from studip_fuse.studipfs.api.aiobase import BaseHTTPClient
from studip_fuse.studipfs.api.aiointerface import Download
log = logging.getLogger(__name__)
async_stat = aiofiles.os.stat
async_utime = aiofiles.os.wrap(os.utime)
class AuthenticatedClientRequest(ClientRequest):
def update_auth(self, auth):
if auth is None:
auth = self.auth
if auth is None:
return
if isinstance(auth, helpers.BasicAuth):
self.headers[hdrs.AUTHORIZATION] = auth.encode()
elif isinstance(auth, OAuth1Client):
url, headers, _ = auth.sign(
str(self.url), str(self.method), None, self.headers
)
async def has_cached_download(self, studip_file, local_dest=None):
if not local_dest:
local_dest = os.path.join(self.cache_dir, studip_file.id)
# check integrity of existing paths (file with id exists, same size, same change date) and reuse them
timestamp = time.mktime(studip_file.changed.timetuple())
try:
stat = await aio_os.stat(local_dest)
return S_ISREG(stat.st_mode) and stat.st_size == studip_file.size and stat.st_mtime == timestamp
except FileNotFoundError:
return False