Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
class InstallRequest(Request):
method: Literal['install']
params: InstallRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_install([split_uri(manager, u) for u in self.params.uris],
self.params.resolution_strategy,
self.params.replace)
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=result.new_pkg, id=self.id)
class UpdateRequestParams(RequestParams):
uris: List[str]
class UpdateRequest(Request):
method: Literal['update']
params: UpdateRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_update([split_uri(manager, u) for u in self.params.uris])
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=result.new_pkg, id=self.id)
class ResolveRequest(Request):
method: Literal['resolve']
params: ResolveRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.resolve([split_uri(manager, u) for u in self.params.uris],
self.params.resolution_strategy)
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=list(result.values()), id=self.id)
class InstallRequestParams(RequestParams):
uris: List[str]
resolution_strategy: Strategies
replace: bool
class Config:
use_enum_values = True
class InstallRequest(Request):
method: Literal['install']
params: InstallRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_install([split_uri(manager, u) for u in self.params.uris],
uris: List[str]
class UpdateRequest(Request):
method: Literal['update']
params: UpdateRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_update([split_uri(manager, u) for u in self.params.uris])
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=result.new_pkg, id=self.id)
class RemoveRequestParams(RequestParams):
uris: List[str]
class RemoveRequest(Request):
method: Literal['remove']
params: RemoveRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_remove([split_uri(manager, u) for u in self.params.uris])
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=None, id=self.id)
params: SetupRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
async def setup() -> Config:
config = Config(**dict(self.params)).write()
setup_logging(config)
manager.finalise(config)
return config
return setup()
def consume_result(self, result: Config) -> SuccessResponse:
return SuccessResponse(result={'config': json_loads(result.json())}, id=self.id)
class ResolveRequestParams(RequestParams):
uris: List[str]
resolution_strategy: Strategies
class Config:
use_enum_values = True
class ResolveRequest(Request):
method: Literal['resolve']
params: ResolveRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.resolve([split_uri(manager, u) for u in self.params.uris],
self.params.resolution_strategy)
uris: List[str]
class RemoveRequest(Request):
method: Literal['remove']
params: RemoveRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
return manager.prep_remove([split_uri(manager, u) for u in self.params.uris])
def consume_result(self, result: Any) -> SuccessResponse:
return SuccessResponse(result=None, id=self.id)
class GetRequestParams(RequestParams):
uris: List[str]
class GetRequest(Request):
method: Literal['get']
params: GetRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
async def get() -> list:
if self.params.uris:
return [manager.get(*split_uri(manager, u)) for u in self.params.uris]
else:
return manager.db.query(Pkg).all()
class Request(Message,
abc.ABC):
id: Union[StrictInt, StrictStr]
method: str
params: RequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
raise NotImplementedError
def consume_result(self, result: Any) -> SuccessResponse:
raise NotImplementedError
class SetupRequestParams(RequestParams):
addon_dir: Path
game_flavour: Literal['retail', 'classic']
class SetupRequest(Request):
method: Literal['setup']
params: SetupRequestParams
def prepare_response(self, manager: WsManager) -> Awaitable:
async def setup() -> Config:
config = Config(**dict(self.params)).write()
setup_logging(config)
manager.finalise(config)
return config