Island: Simplify logic for checking duplicate URLs

This commit is contained in:
Mike Salvatore 2022-05-23 07:54:49 -04:00 committed by vakarisz
parent 0344ee1a7c
commit 108c86d56c
1 changed files with 8 additions and 16 deletions

View File

@ -1,12 +1,11 @@
import os import os
import uuid import uuid
from datetime import timedelta from datetime import timedelta
from typing import Sequence, Type from typing import Iterable, Type
import flask_restful import flask_restful
from flask import Flask, Response, send_from_directory from flask import Flask, Response, send_from_directory
from werkzeug.exceptions import NotFound from werkzeug.exceptions import NotFound
from werkzeug.routing import Map
from common import DIContainer from common import DIContainer
from monkey_island.cc.database import database, mongo from monkey_island.cc.database import database, mongo
@ -117,29 +116,22 @@ class FlaskDIWrapper:
def __init__(self, api: flask_restful.Api, container: DIContainer): def __init__(self, api: flask_restful.Api, container: DIContainer):
self._api = api self._api = api
self._container = container self._container = container
self._registered_urls = set()
def add_resource(self, resource: Type[IResource]): def add_resource(self, resource: Type[IResource]):
self._register_unique_urls(resource.urls)
dependencies = self._container.resolve_dependencies(resource) dependencies = self._container.resolve_dependencies(resource)
FlaskDIWrapper._check_for_duplicate_urls(self._api.app.url_map, resource.urls)
self._api.add_resource(resource, *resource.urls, resource_class_args=dependencies) self._api.add_resource(resource, *resource.urls, resource_class_args=dependencies)
@staticmethod def _register_unique_urls(self, urls: Iterable[str]):
def _check_for_duplicate_urls(url_map: Map, urls: Sequence[str]): for url in map(lambda x: x.rstrip("/"), urls):
for url in urls: if url in self._registered_urls:
if FlaskDIWrapper._is_url_added(url_map, url):
raise FlaskDIWrapper.URLAlreadyExistsError( raise FlaskDIWrapper.URLAlreadyExistsError(
f"URL {url} has already been registered!" f"URL {url} has already been registered!"
) )
@staticmethod self._registered_urls.add(url)
def _is_url_added(url_map: Map, url_to_add: str) -> bool:
return bool(
[
registered_url
for registered_url in url_map.iter_rules()
if str(registered_url).strip("/") == url_to_add.strip("/")
]
)
def init_api_resources(api: FlaskDIWrapper): def init_api_resources(api: FlaskDIWrapper):