Python Examples of aiohttp.web.FileResponse (2023)

How To Fix Java Errors for Minecraft

def get_tails_file(request: web.BaseRequest) -> web.FileResponse: """ Request handler to download the tails file of the revocation registry. Args: request: aiohttp request object Returns: The tails file in FileResponse """ context = request.app["request_context"] registry_id = request.match_info["rev_reg_id"] try: revoc = IndyRevocation(context) revoc_registry = await revoc.get_issuer_rev_reg_record(registry_id) except StorageNotFoundError as err: raise web.HTTPNotFound(reason=err.roll_up) from err return web.FileResponse(path=revoc_registry.tails_local_path, status=200) 

def setup_router(self): self.app.router.add_get('/get/proxy/', self.get_proxies_html) self.app.router.add_get('/get/proxy_count_item/', self.get_proxy_count_items_html) self.app.router.add_get('/get/number_of_proxies_to_process/', self.get_number_of_proxies_to_process_html) self.app.router.add_get('/get/processor_proxies_queue_size/', self.get_processor_proxies_queue_size_html) self.app.router.add_get('/get/collector_state/', self.get_collector_state_html) self.app.router.add_get('/get/best/http/proxy/', self.get_best_http_proxy) # self.app.router.add_get('/{tail:.*}', self.default_route) # # async def default_route(self, request: aiohttp.ClientRequest): # path = request.path # if path == '/': # path = '/index.html' # # if re.match(r'^(/([a-zA-Z0-9_]+(\.[a-zA-Z0-9]+)*)?)+$', path): # try: # path = os.path.join('./server/frontend/angular/dist', path) # with open(path, 'r'): # return web.FileResponse(path) # except (FileNotFoundError, IsADirectoryError): # pass # # return web.FileResponse('./server/frontend/angular/dist/index.html') 

def static_request_handler(cls: Any, obj: Any, context: Dict, func: Any, path: str, base_url: str, ignore_logging: Union[bool, List[int], Tuple[int]] = False) -> Any: if '?P<filename>' not in base_url: pattern = r'^{}(?P<filename>.+?)$'.format(re.sub(r'\$$', '', re.sub(r'^\^?(.*)$', r'\1', base_url))) else: pattern = r'^{}$'.format(re.sub(r'\$$', '', re.sub(r'^\^?(.*)$', r'\1', base_url))) compiled_pattern = re.compile(pattern) if path.startswith('/'): path = os.path.dirname(path) else: path = '{}/{}'.format(os.path.dirname(context.get('context', {}).get('_service_file_path')), path) if not path.endswith('/'): path = '{}/'.format(path) async def handler(request: web.Request) -> web.Response: result = compiled_pattern.match(request.path) filename = result.groupdict()['filename'] if result else '' filepath = '{}{}'.format(path, filename) try: if os.path.commonprefix((os.path.realpath(filepath), os.path.realpath(path))) != os.path.realpath(path) or os.path.isdir(filepath) or not os.path.exists(filepath): raise web.HTTPNotFound() # type: ignore pathlib.Path(filepath).open('r') response = FileResponse(path=filepath, # type: ignore chunk_size=256 * 1024) # type: web.Response return response except PermissionError as e: raise web.HTTPForbidden() # type: ignore route_context = {'ignore_logging': ignore_logging} context['_http_routes'] = context.get('_http_routes', []) context['_http_routes'].append(('GET', pattern, handler, route_context)) start_func = cls.start_server(obj, context) return (await start_func) if start_func else None 

def resolve_response(value: Union[str, bytes, Dict, List, Tuple, web.Response, Response], request: Optional[web.Request] = None, context: Dict = None, status_code: Optional[Union[str, int]] = None, default_content_type: Optional[str] = None, default_charset: Optional[str] = None) -> web.Response: if not context: context = {} if isinstance(value, Response): return value.get_aiohttp_response(context, default_content_type=default_content_type, default_charset=default_charset) if isinstance(value, web.FileResponse): return value status = int(status_code) if status_code else (request is not None and request._cache.get('error_status_code', 200)) or 200 headers = None if isinstance(value, dict): body = value.get('body') _status = value.get('status') # type: Optional[SupportsInt] if _status and isinstance(_status, (int, str, bytes)): status = int(_status) _returned_headers = value.get('headers') if _returned_headers: returned_headers = _returned_headers # type: Union[Mapping[str, Any], Iterable[Tuple[str, Any]]] headers = CIMultiDict(returned_headers) elif isinstance(value, list) or isinstance(value, tuple): _status = value[0] if _status and isinstance(_status, (int, str, bytes)): status = int(_status) body = value[1] if len(value) > 2: returned_headers = value[2] headers = CIMultiDict(returned_headers) elif isinstance(value, web.Response): return value else: if value is None: value = '' body = value return Response(body=body, status=status, headers=headers, content_type=default_content_type, charset=default_charset).get_aiohttp_response(context) 

def download_single(request: web.Request, params: Any, row: VFolderRow) -> web.StreamResponse: folder_name = request.match_info['name'] access_key = request['keypair']['access_key'] fn = params['file'] log.info('VFOLDER.DOWNLOAD_SINGLE (ak:{}, vf:{}, path:{})', access_key, folder_name, fn) folder_path = get_folder_hostpath(row, request.app) try: file_path = (folder_path / fn).resolve() file_path.relative_to(folder_path) if not file_path.exists(): raise FileNotFoundError except (ValueError, FileNotFoundError): raise GenericNotFound('The file is not found.') if not file_path.is_file(): if params['archive']: # Download directory as an archive when archive param is set. return await download_directory_as_archive(request, file_path) else: raise InvalidAPIParameters('The file is not a regular file.') if request.method == 'HEAD': return web.Response(status=200, headers={ hdrs.ACCEPT_RANGES: 'bytes', hdrs.CONTENT_LENGTH: str(file_path.stat().st_size), }) ascii_filename = file_path.name.encode('ascii', errors='ignore').decode('ascii').replace('"', r'\"') encoded_filename = urllib.parse.quote(file_path.name, encoding='utf-8') return web.FileResponse(file_path, headers={ hdrs.CONTENT_TYPE: "application/octet-stream", hdrs.CONTENT_DISPOSITION: " ".join([ "attachment;" f"filename=\"{ascii_filename}\";", # RFC-2616 sec2.2 f"filename*=UTF-8''{encoded_filename}", # RFC-5987 ]) }) 

def get_task_logs(request: web.Request, params: Any) -> web.StreamResponse: log.info('GET_TASK_LOG (ak:{}, k:{})', request['keypair']['access_key'], params['kernel_id']) domain_name = request['user']['domain_name'] user_role = request['user']['role'] user_uuid = request['user']['uuid'] raw_kernel_id = params['kernel_id'].hex mount_prefix = await request.app['config_server'].get('volumes/_mount') fs_prefix = await request.app['config_server'].get('volumes/_fsprefix') async with request.app['dbpool'].acquire() as conn, conn.begin(): matched_vfolders = await query_accessible_vfolders( conn, user_uuid, user_role=user_role, domain_name=domain_name, allowed_vfolder_types=['user'], extra_vf_conds=(vfolders.c.name == '.logs')) if not matched_vfolders: raise GenericNotFound('You do not have ".logs" vfolder for persistent task logs.') log_vfolder = matched_vfolders[0] log_path = ( Path(mount_prefix) / log_vfolder['host'] / Path(fs_prefix.lstrip('/')) / log_vfolder['id'].hex / 'task' / raw_kernel_id[:2] / raw_kernel_id[2:4] / f'{raw_kernel_id[4:]}.log' ) def check_file(): if not log_path.is_file(): raise GenericNotFound('The requested log file or the task was not found.') try: with open(log_path, 'rb'): pass except IOError: raise GenericNotFound('The requested log file is not readable.') loop = current_loop() await loop.run_in_executor(None, check_file) return web.FileResponse(log_path, headers={ hdrs.CONTENT_TYPE: "text/plain", }) 

def redirect_index(_: web.Request) -> web.FileResponse: return web.FileResponse(pkg_resources.resource_filename("mautrix_hangouts", "web/static/login-redirect.html")) 

def get_index(_: web.Request) -> web.FileResponse: return web.FileResponse(pkg_resources.resource_filename("mautrix_hangouts", "web/static/login.html")) 

def handle_upload_global_file(self, request): try: fileID = request.match_info.get("id", "") return web.FileResponse(os.path.join(bumper.bumper_dir,"bumper","web","images","robotvac_image.jpg")) except Exception as e: logging.exception("{}".format(e)) 

def handle_pimFile(self, request): try: fileID = request.match_info.get("id", "") return web.FileResponse(os.path.join(bumper.bumper_dir,"bumper","web","images","robotvac_image.jpg")) except Exception as e: logging.exception("{}".format(e)) 

def download_song(request: web.Request) -> web.FileResponse: """GET /api/install/song/{song_id} Description: Download a song by its ID. Example: link: /api/install/song/905110 Returns: 200: Found song; 400: Invalid type; 404: Failed to find the song. Return Type: audio/mpeg """ song_id = int(request.match_info["song_id"]) path = ROOT_PATH / f"song-{song_id}.mp3" if path.exists(): return web.FileResponse(path) song = await request.app.client.get_ng_song(song_id) await song.download(file=path) request.loop.create_task(delete_after(60, path)) return web.FileResponse(path) 

def generate_icons(request: web.Request) -> web.Response: query = multidict.CIMultiDict(request.query) color_1 = color_from_hex(query.pop("color_1", "0x00ff00")) color_2 = color_from_hex(query.pop("color_2", "0x00ffff")) glow_outline = str_to_bool(query.pop("glow_outline", "false")) error_on_not_found = str_to_bool(query.pop("error_on_not_found", "false")) settings = f"color_1={color_1}$color_2={color_2}$glow_outline={glow_outline}".lower() types = "$".join(f"{key}={value}".lower() for key, value in query.items()) name = f"[{settings}]({types}).png" path = ROOT_PATH / name if path.exists(): return web.FileResponse(path) images = [ await gd.utils.run_blocking_io( gd.factory.generate, icon_type=gd.IconType.from_value(icon_type), icon_id=int(icon_id), color_1=color_1, color_2=color_2, glow_outline=glow_outline, error_on_not_found=error_on_not_found, ) for icon_type, icon_id in query.items() ] if not images: raise Warning("No types were generated.") image = await gd.utils.run_blocking_io(gd.icon_factory.connect_images, images) image.save(path) request.loop.create_task(delete_after(5, path)) return web.FileResponse(path) 

def test_funnel(tmp_path, aiohttp_client): tmp_file = tmp_path / 'test' data = bytes(getrandbits(8) for _ in range(16)) tmp_file.write_bytes(data) async def serve_file(request): # FileResponse supports range requests. return web.FileResponse(tmp_file) app = web.Application() app.router.add_get('/', serve_file) session = await aiohttp_client(app) async def test(block_size, piece_size): r = HttpRange(0, len(data) - 1) buf = BytesIO() async with Funnel( url='/', range=r, session=session, block_size=block_size, piece_size=piece_size) as funnel: async for block in funnel: buf.write(block) assert buf.getvalue() == data tests = [] for block_size in range(1, len(data) + 1): for piece_size in range(1, block_size + 1): tests.append(asyncio.create_task(test(block_size, piece_size))) await asyncio.gather(*tests) 

def index(request): return web.FileResponse(ROOT / 'index.html') 

def favicon(request): return web.FileResponse("static/favicon.ico") 

def handle(request): global filename global PAYLOAD if request.headers["Host"] is not None: if "downloads.dell.com" in request.headers["Host"]: print("[+] Exploit binary requested.") return web.FileResponse(filename) elif "dell.com" in request.headers["Host"]: print("[+] Exploit payload requested.") return web.Response(text=PAYLOAD, headers={'Content-Type': 'text/html'}) redirect_url = "http://dellrce.dell.com" return web.HTTPFound(redirect_url) 

def _insert_footer(self, response): if not isinstance(response, FileResponse) or not self._add_tail_snippet: return response filepath = response._path ct, encoding = mimetypes.guess_type(str(response._path)) if ct != 'text/html': return response with filepath.open('rb') as f: body = f.read() + LIVE_RELOAD_LOCAL_SNIPPET resp = Response(body=body, content_type='text/html') resp.last_modified = filepath.stat().st_mtime return resp 

def httpserver_index_handler(self, request): return web.FileResponse(self._web_dir + '/index.html') 

def handleWeb(req): # serve all statics from web folder ressource = req.match_info.get("path", "") if ressource == "" or ressource.endswith("/"): ressource += "index.html" if ressource.lower().endswith((".html", ".htm")): name = getname(ressource) if name in currents: html = currents[name]._render(path(os.path.dirname(ressource))) if html: # the instance render its own html, go with it if re.findall(r"""<.*script.*src.*=.*['"]wuy.js['"].*>""", html): return web.Response(status=200, body=html, content_type="text/html") else: return web.Response( status=200, body='<script src="wuy.js"></script>\n' + html, content_type="text/html", ) # classic serve static file or 404 file = path(os.path.join("web", ressource)) if os.path.isfile(file): wlog("- serve static file", file) return web.FileResponse(file) else: wreq = Request(req) if req.body_exists: # override body wreq.body = await req.read() for name, instance in currents.items(): ret = instance.request(wreq) if ret is not None: r = await ret if asyncio.iscoroutine(ret) else ret if r is not None: if type(r) != Response: r = Response(status=200, content=r) wlog("- serve dynamic via", name, r) return web.Response( status=r.status, body=r.content, headers=r.headers ) wlog("! 404 on", file) return web.Response(status=404, body="file not found") 

def request_handler(cls: Any, obj: Any, context: Dict, func: Any, method: str, url: str, ignore_logging: Union[bool, List[int], Tuple[int]] = False, pre_handler_func: Optional[Callable] = None) -> Any: pattern = r'^{}$'.format(re.sub(r'\$$', '', re.sub(r'^\^?(.*)$', r'\1', url))) compiled_pattern = re.compile(pattern) default_content_type = context.get('options', {}).get('http', {}).get('content_type', 'text/plain') default_charset = context.get('options', {}).get('http', {}).get('charset', 'utf-8') if default_content_type is not None and ";" in default_content_type: # for backwards compability try: default_charset = str([v for v in default_content_type.split(';') if 'charset=' in v][0]).replace('charset=', '').strip() default_content_type = str([v for v in default_content_type.split(';')][0]).strip() except IndexError: pass async def handler(request: web.Request) -> Union[web.Response, web.FileResponse]: result = compiled_pattern.match(request.path) values = inspect.getfullargspec(func) kwargs = {k: values.defaults[i] for i, k in enumerate(values.args[len(values.args) - len(values.defaults):])} if values.defaults else {} if result: for k, v in result.groupdict().items(): kwargs[k] = v @functools.wraps(func) async def routine_func(*a: Any, **kw: Any) -> Union[str, bytes, Dict, List, Tuple, web.Response, Response]: routine = func(*(obj, request, *a), **merge_dicts(kwargs, kw)) return_value = (await routine) if isinstance(routine, Awaitable) else routine # type: Union[str, bytes, Dict, List, Tuple, web.Response, Response] return return_value if not context.get('_http_accept_new_requests'): raise web.HTTPServiceUnavailable() if pre_handler_func: await pre_handler_func(obj, request) return_value = await execute_middlewares(func, routine_func, context.get('http_middleware', []), *(obj, request)) response = await resolve_response(return_value, request=request, context=context, default_content_type=default_content_type, default_charset=default_charset) return response context['_http_routes'] = context.get('_http_routes', []) route_context = {'ignore_logging': ignore_logging} if isinstance(method, list) or isinstance(method, tuple): for m in method: context['_http_routes'].append((m.upper(), pattern, handler, route_context)) else: context['_http_routes'].append((method.upper(), pattern, handler, route_context)) start_func = cls.start_server(obj, context) return (await start_func) if start_func else None 

def download_with_token(request) -> web.StreamResponse: try: secret = request.app['config']['manager']['secret'] token = request.query.get('token', '') params = jwt.decode(token, secret, algorithms=['HS256']) except jwt.PyJWTError: log.exception('jwt error while parsing "{}"', token) raise InvalidAPIParameters('Could not validate the download token.') iv = t.Dict({ t.Key('file'): t.String, t.Key('host'): t.String, t.Key('id'): t.String, t.Key('exp'): t.Int, t.Key('archive', default=False): t.Bool | t.Null, }) params = iv.check(params) fn = params['file'] log.info('VFOLDER.DOWNLOAD_WITH_TOKEN (token:{}, path:{})', token, fn) dbpool = request.app['dbpool'] async with dbpool.acquire() as conn: query = (sa.select([vfolders.c.unmanaged_path]) .select_from(vfolders) .where(vfolders.c.id == params['id']) .limit(1)) unmanaged_path = await conn.scalar(query) if unmanaged_path: folder_path = Path(unmanaged_path) else: folder_path = (request.app['VFOLDER_MOUNT'] / params['host'] / request.app['VFOLDER_FSPREFIX'] / params['id']) try: file_path = (folder_path / fn).resolve() file_path.relative_to(folder_path) if not file_path.exists(): raise FileNotFoundError except (ValueError, FileNotFoundError): raise InvalidAPIParameters('The file is not found.') if not file_path.is_file(): if params['archive']: # Download directory as an archive when archive param is set. return await download_directory_as_archive(request, file_path) else: raise InvalidAPIParameters('The file is not a regular file.') if request.method == 'HEAD': return web.Response(status=200, headers={ hdrs.ACCEPT_RANGES: 'bytes', hdrs.CONTENT_LENGTH: str(file_path.stat().st_size), }) ascii_filename = file_path.name.encode('ascii', errors='ignore').decode('ascii').replace('"', r'\"') encoded_filename = urllib.parse.quote(file_path.name, encoding='utf-8') return web.FileResponse(file_path, headers={ hdrs.CONTENT_TYPE: "application/octet-stream", hdrs.CONTENT_DISPOSITION: " ".join([ "attachment;" f"filename=\"{ascii_filename}\";", # RFC-2616 sec2.2 f"filename*=UTF-8''{encoded_filename}", # RFC-5987 ]) }) 

def get_icons(request: web.Request) -> web.Response: """GET /api/icons/{type:(all|main|cube|ship|ball|ufo|wave|robot|spider)}/{query} Description: Generate icon of the user given by query. Example: link: /api/icons/all/5509312?id=true id: true Parameters: id: Whether to interpret "query" as AccountID or Name/PlayerID; Returns: 200: Image with generated icons; 404: Could not find requested user. Return Type: image/png """ icon_type = request.match_info["type"] query = request.match_info["query"] is_id = str_to_bool(request.query.get("id", "false")) if is_id: user = await request.app.client.get_user(int(query)) else: user = await request.app.client.search_user(query) path = ROOT_PATH / f"icons-{icon_type}-{user.account_id}.png" if path.exists(): return web.FileResponse(path) if icon_type == "main": icon_type = user.icon_set.main_type.name.lower() if icon_type == "all": image = await user.icon_set.generate_full(as_image=True) else: image = await user.icon_set.generate(icon_type, as_image=True) image.save(path) request.loop.create_task(delete_after(5, path)) return web.FileResponse(path) 

def get_file_response(requested_file): """Get file.""" hacs = get_hacs() if requested_file.startswith("frontend-"): if hacs.configuration.debug: servefile = await hacs.hass.async_add_executor_job(locate_debug_gz) hacs.logger.debug("Serving DEBUG frontend") else: servefile = await hacs.hass.async_add_executor_job(locate_gz) if os.path.exists(servefile): return web.FileResponse(servefile) elif requested_file == "iconset.js": return web.FileResponse( f"{hacs.system.config_path}/custom_components/hacs/iconset.js" ) try: if requested_file.startswith("themes"): file = f"{hacs.system.config_path}/{requested_file}" else: file = f"{hacs.system.config_path}/www/community/{requested_file}" # Serve .gz if it exist if os.path.exists(file + ".gz"): file += ".gz" if os.path.exists(file): hacs.logger.debug("Serving {} from {}".format(requested_file, file)) response = web.FileResponse(file) response.headers["Cache-Control"] = "no-store, max-age=0" response.headers["Pragma"] = "no-store" return response else: hacs.logger.error(f"Tried to serve up '{file}' but it does not exist") except Exception as error: # pylint: disable=broad-except hacs.logger.debug( "there was an issue trying to serve {} - {}".format(requested_file, error) ) return web.Response(status=404)