ED/EDMarketConnector
1
0
Fork 0
mirror of https://github.com/EDCD/EDMarketConnector.git synced 2025-04-13 07:47:14 +03:00
Code Issues Projects Releases Wiki Activity

Typing of return values from killswitch.check_killswitch()

Browse Source
This commit is contained in:
aussig 2022-12-22 17:39:20 +00:00
parent 9e17c46b25
commit 3e295db061
6 changed files with 66 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
EDMarketConnector.py
View File

@ -16,7 +16,7 @@ from builtins import object, str
from os import chdir, environ
from os.path import dirname, join
from time import localtime, strftime, time
from typing import TYPE_CHECKING, Literal, Optional, Tuple, Union
from typing import TYPE_CHECKING, Any, Dict, Literal, Optional, Tuple, Union
# Have this as early as possible for people running EDMarketConnector.exe
# from cmd.exe or a bat file or similar. Else they might not be in the correct
@ -916,7 +916,10 @@ class AppWindow(object):
def login(self):
"""Initiate CAPI/Frontier login and set other necessary state."""
should_return, __ = killswitch.check_killswitch('capi.auth', {})
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.auth', {})
if should_return:
logger.warning('capi.auth has been disabled via killswitch. Returning.')
self.status['text'] = 'CAPI auth disabled by killswitch'
@ -1007,7 +1010,10 @@ class AppWindow(object):
:param event: Tk generated event details.
"""
logger.trace_if('capi.worker', 'Begin')
should_return, __ = killswitch.check_killswitch('capi.auth', {})
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.auth', {})
if should_return:
logger.warning('capi.auth has been disabled via killswitch. Returning.')
self.status['text'] = 'CAPI auth disabled by killswitch'
@ -1093,7 +1099,10 @@ class AppWindow(object):
:param event: Tk generated event details.
"""
logger.trace_if('capi.worker', 'Begin')
should_return, __ = killswitch.check_killswitch('capi.request.fleetcarrier', {})
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request.fleetcarrier', {})
if should_return:
logger.warning('capi.fleetcarrier has been disabled via killswitch. Returning.')
self.status['text'] = 'CAPI fleetcarrier disabled by killswitch'
@ -1301,7 +1310,10 @@ class AppWindow(object):
if err:
play_bad = True
should_return, __ = killswitch.check_killswitch('capi.request./market', {})
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request./market', {})
if should_return:
logger.warning("capi.request./market has been disabled by killswitch. Returning.")
@ -1555,13 +1567,16 @@ class AppWindow(object):
elif entry['event'] == 'Disembark' and entry.get('Taxi') and entry.get('OnStation'):
auto_update = True
should_return: bool
new_data: Dict[str, Any] = {}
if auto_update:
should_return, __ = killswitch.check_killswitch('capi.auth', {})
should_return, new_data = killswitch.check_killswitch('capi.auth', {})
if not should_return:
self.w.after(int(SERVER_RETRY * 1000), self.capi_request_data)
if entry['event'] in ('CarrierBuy', 'CarrierStats'):
should_return, __ = killswitch.check_killswitch('capi.request.fleetcarrier', {})
should_return, new_data = killswitch.check_killswitch('capi.request.fleetcarrier', {})
if not should_return:
self.w.after(int(SERVER_RETRY * 1000), self.capi_request_fleetcarrier_data)

11
companion.py
View File

@ -328,6 +328,9 @@ class Auth(object):
"""
logger.debug(f'Trying for "{self.cmdr}"')
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.auth', {})
if should_return:
logger.warning('capi.auth has been disabled via killswitch. Returning.')
@ -667,6 +670,9 @@ class Session(object):
:return: True if login succeeded, False if re-authorization initiated.
"""
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.auth', {})
if should_return:
logger.warning('capi.auth has been disabled via killswitch. Returning.')
@ -785,6 +791,9 @@ class Session(object):
:return: The resulting CAPI data, of type CAPIData.
"""
capi_data: CAPIData = CAPIData()
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request.' + capi_endpoint, {})
if should_return:
logger.warning(f"capi.request.{capi_endpoint} has been disabled by killswitch. Returning.")
@ -962,7 +971,7 @@ class Session(object):
elif query.endpoint == self.FRONTIER_CAPI_PATH_FLEETCARRIER:
capi_data = capi_single_query(query.capi_host, self.FRONTIER_CAPI_PATH_FLEETCARRIER,
timeout=capi_fleetcarrier_requests_timeout)
timeout=capi_fleetcarrier_requests_timeout)
else:
capi_data = capi_single_query(query.capi_host, self.FRONTIER_CAPI_PATH_PROFILE)

5
plugins/eddb.py
View File

@ -42,7 +42,7 @@
# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $#
# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $# ! $#
import tkinter
from typing import TYPE_CHECKING, Any, Mapping, MutableMapping, Optional
from typing import TYPE_CHECKING, Any, Dict, Mapping, MutableMapping, Optional
import requests
@ -168,6 +168,9 @@ def journal_entry( # noqa: CCR001
:param state: `monitor.state`
:return: None if no error, else an error string.
"""
should_return: bool
new_entry: Dict[str, Any] = {}
should_return, new_entry = killswitch.check_killswitch('plugins.eddb.journal', entry)
if should_return:
# LANG: Journal Processing disabled due to an active killswitch

15
plugins/eddn.py
View File

@ -402,6 +402,9 @@ class EDDNSender:
:return: `True` for "now remove this message from the queue"
"""
logger.trace_if("plugin.eddn.send", "Sending message")
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('plugins.eddn.send', json.loads(msg))
if should_return:
logger.warning('eddn.send has been disabled via killswitch. Returning.')
@ -635,6 +638,9 @@ class EDDN:
:param data: a dict containing the starport data
:param is_beta: whether or not we're currently in beta mode
"""
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request./market', {})
if should_return:
logger.warning("capi.request./market has been disabled by killswitch. Returning.")
@ -766,6 +772,9 @@ class EDDN:
:param data: dict containing the outfitting data
:param is_beta: whether or not we're currently in beta mode
"""
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request./shipyard', {})
if should_return:
logger.warning("capi.request./shipyard has been disabled by killswitch. Returning.")
@ -832,6 +841,9 @@ class EDDN:
:param data: dict containing the shipyard data
:param is_beta: whether or not we are in beta mode
"""
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('capi.request./shipyard', {})
if should_return:
logger.warning("capi.request./shipyard has been disabled by killswitch. Returning.")
@ -2164,6 +2176,9 @@ def journal_entry( # noqa: C901, CCR001
:param state: `dict` - Current `monitor.state` data.
:return: `str` - Error message, or `None` if no errors.
"""
should_return: bool
new_data: Dict[str, Any] = {}
should_return, new_data = killswitch.check_killswitch('plugins.eddn.journal', entry)
if should_return:
plug.show_error(_('EDDN journal handler disabled. See Log.')) # LANG: Killswitch disabled EDDN

11
plugins/edsm.py
View File

@ -38,7 +38,7 @@ from datetime import datetime, timedelta, timezone
from queue import Queue
from threading import Thread
from time import sleep
from typing import TYPE_CHECKING, Any, List, Literal, Mapping, MutableMapping, Optional, Set, Tuple, Union, cast
from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, MutableMapping, Optional, Set, Tuple, Union, cast
import requests
@ -479,6 +479,9 @@ def journal_entry( # noqa: C901, CCR001
:param state: `monitor.state`
:return: None if no error, else an error string.
"""
should_return: bool
new_entry: Dict[str, Any] = {}
should_return, new_entry = killswitch.check_killswitch('plugins.edsm.journal', entry, logger)
if should_return:
# LANG: EDSM plugin - Journal handling disabled by killswitch
@ -759,6 +762,9 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
retrying = 0
while retrying < 3:
should_skip: bool
new_item: Dict[str, Any] = {}
should_skip, new_item = killswitch.check_killswitch(
'plugins.edsm.worker',
item if item is not None else cast(Tuple[str, Mapping[str, Any]], ("", {})),
@ -795,6 +801,9 @@ def worker() -> None: # noqa: CCR001 C901 # Cant be broken up currently
# drop events if required by killswitch
new_pending = []
for e in pending:
skip: bool
new: Dict[str, Any] = {}
skip, new = killswitch.check_killswitch(f'plugin.edsm.worker.{e["event"]}', e, logger)
if skip:
continue

5
plugins/inara.py
View File

@ -385,6 +385,9 @@ def journal_entry( # noqa: C901, CCR001
return ''
should_return: bool
new_entry: Dict[str, Any] = {}
should_return, new_entry = killswitch.check_killswitch('plugins.inara.journal', entry, logger)
if should_return:
plug.show_error(_('Inara disabled. See Log.')) # LANG: INARA support disabled via killswitch
@ -1536,6 +1539,8 @@ def new_add_event(
def clean_event_list(event_list: List[Event]) -> List[Event]:
"""Check for killswitched events and remove or modify them as requested."""
out = []
bad: bool
new_event: Dict[str, Any] = {}
for e in event_list:
bad, new_event = killswitch.check_killswitch(f'plugins.inara.worker.{e.name}', e.data, logger)