Basic functional implemented: can update squadron with specified id

FAPI/BearerManager.py

@@ -0,0 +1,39 @@
+from enum import Enum
+from loguru import logger
+import os
+import requests
+
+
+class BearerManager:
+    class Endpoints(Enum):
+        RANDOM = '/random_token'
+
+    def __init__(self, demb_capi_auth: str, base_address: str):
+
+        self.base_address = base_address
+        self.demb_capi_auth = demb_capi_auth
+
+    def get_random_bearer(self) -> str:
+        """Gets bearer token from capi.demb.design (companion-api project)
+
+        :return: bearer token as str
+        """
+
+        bearer_request: requests.Response = self._request(self.Endpoints.RANDOM)
+
+        try:
+            bearer: str = bearer_request.json()['access_token']
+
+        except Exception as e:
+            logger.exception(f'Unable to parse capi.demb.design answer\nrequested: {bearer_request.url!r}\n'
+                             f'code: {bearer_request.status_code!r}\nresponse: {bearer_request.content!r}', exc_info=e)
+            raise e
+
+        return bearer
+
+    def _request(self, _endpoint: Endpoints) -> requests.Response:
+        endpoint = self.base_address + _endpoint.value
+        return requests.get(url=endpoint, headers={'auth': self.demb_capi_auth})
+
+
+bearer_manager = BearerManager(os.environ['DEMB_CAPI_AUTH'], 'https://capi.demb.design')

FAPI/Converters.py

@@ -0,0 +1,3 @@
+def dehexify(hex_str: str) -> str:
+    """Converts string with hex chars to string"""
+    return bytes.fromhex(hex_str).decode('utf-8')

FAPI/Exceptions.py

@@ -0,0 +1,6 @@
+class FAPIDownForMaintenance(Exception):
+    pass
+
+
+class FAPIUnknownStatusCode(Exception):
+    pass

FAPI/Mappings.py

@@ -0,0 +1,45 @@
+info_request_mapping = {
+    "id": "squad_id",
+    "name": "name",
+    "tag": "tag",
+    "ownerId": "owner_id",
+    "ownerName": "owner_name",
+    "platform": "platform",
+    "created": "created",
+    "created_ts": "created_ts",
+    "acceptingNewMembers": "accepting_new_members",
+    "powerId": "power_id",
+    "powerName": "power_name",
+    "superpowerId": "superpower_id",
+    "superpowerName": "superpower_name",
+    "factionId": "faction_id",
+    "factionName": "faction_name",
+    "deleteAfter": "delete_after",
+    "userTags": "user_tags",
+    "memberCount": "member_count",
+    "onlineCount": "online_count",
+    "pendingCount": "pending_count",
+    "publicComms": "public_comms",
+    "publicCommsOverride": "public_comms_override",
+    "publicCommsAvailable": "public_comms_available"
+}
+
+news_request_mapping = {
+    'id': 'news_id'
+}
+
+
+def perform_mapping(mapping: dict, dict_to_map: dict) -> dict:
+    for key in (list(dict_to_map.keys())):
+        if key in mapping:
+            dict_to_map[mapping[key]] = dict_to_map.pop(key)
+
+    return dict_to_map
+
+
+def perform_info_mapping(info_data: dict) -> dict:
+    return perform_mapping(info_request_mapping, info_data)
+
+
+def perform_news_mapping(news_data: dict) -> dict:
+    return perform_mapping(news_request_mapping, news_data)

FAPI/Queries.py

@@ -0,0 +1,88 @@
+from .Mappings import perform_info_mapping, perform_news_mapping
+from .Exceptions import *
+from .Requester import *
+from loguru import logger
+from . import Converters
+import json
+
+
+def get_squad_news(squad_id) -> dict | None:
+    """
+    Returns news of squadron with specified id or None if squadron doesn't exists or there is no news
+
+    :param squad_id:
+    :return:
+    """
+
+    news_request: requests.Response = request(BASE_URL + NEWS_ENDPOINT, params={'squadronId': squad_id})
+    if news_request.status_code != 200:  # must not happen
+        logger.warning(f'Got not 200 status code on requesting news, content: {news_request.content}, '
+                       f'code: {news_request.status_code}')
+
+    squad_news: dict = news_request.json()['squadron']
+
+    if isinstance(squad_news, list):  # check squadron 2517 for example 0_0
+        # squadron have no public statements
+        return None
+
+    elif 'id' not in squad_news.keys():  # squadron doesn't FDEV
+        return None
+
+    else:
+        if 'public_statements' in squad_news.keys() and len(squad_news['public_statements']) > 0:
+            return perform_news_mapping(squad_news['public_statements'][0])
+
+
+def get_squad_info(squad_id: int) -> dict | None:
+    """Returns information about squadron with specified id or None if squadrons doesn't exists on FDEV side
+
+    :param squad_id: id of squad to update/insert
+    :return: dict with state of squadron or None if squad not found
+    """
+
+    """
+    How it should works? 
+    Request squad's info
+    
+    if squad exists FDEV
+        return squadron state
+    
+    if squad doesn't exists FDEV        
+       return None
+    """
+    """
+    if DB.db.execute(SQLRequests.squad_deleted, {'squad_id': squad_id}) == 1:
+        # we have it deleted in our DB
+        logger.debug(f'squad {squad_id} is marked as deleted in our DB, returning False')
+        return None
+    """
+
+    squad_request: requests.Response = request(BASE_URL + INFO_ENDPOINT, params={'squadronId': squad_id})
+
+    if squad_request.status_code == 200:  # squad exists FDEV
+        squad_request_json: dict = squad_request.json()['squadron']
+        squad_request_json['ownerName'] = Converters.dehexify(squad_request_json['ownerName'])  # normalize value
+        squad_request_json['userTags'] = json.dumps(squad_request_json['userTags'])
+        squad_request_json = perform_info_mapping(squad_request_json)
+
+        """
+        with DB.db:
+            operation_id = DB.allocate_operation_id(squad_id)
+            squad_request_json['operation_id'] = operation_id
+            DB.db.execute(SQLRequests.insert_info, squad_request_json)
+         """
+
+        return squad_request_json
+
+    elif squad_request.status_code == 404:  # squad doesn't exists FDEV
+        """
+        if not suppress_absence:
+            with DB.db:
+                operation_id = DB.allocate_operation_id(squad_id)
+                DB.db.execute(SQLRequests.delete_squadron, {'operation_id': operation_id, 'squad_id': squad_id})
+        """
+        return None
+
+    else:  # any other codes (except 418, that one handles in authed_request), never should happen
+        logger.warning(f'Unknown squad info status_code: {squad_request.status_code}, content: {squad_request.content}')
+        raise FAPIUnknownStatusCode(f'Status code: {squad_request.status_code}, content: {squad_request.content}')

FAPI/Requester.py

@@ -0,0 +1,30 @@
+import requests
+from .BearerManager import bearer_manager
+from loguru import logger
+from . import Exceptions
+
+
+"""
+Functions to perform queries to FDEV
+"""
+
+
+BASE_URL = 'https://api.orerve.net/2.0/website/squadron/'
+INFO_ENDPOINT = 'info'
+NEWS_ENDPOINT = 'news/list'
+
+
+def request(url: str, method: str = 'get', **kwargs) -> requests.Response:
+    _request: requests.Response = requests.request(
+        method=method,
+        url=url,
+        headers={'Authorization': f'Bearer {bearer_manager.get_random_bearer()}'},
+        **kwargs
+    )
+
+    if _request.status_code == 418:  # FAPI is on maintenance
+        logger.warning(f'{method.upper()} {_request.url} returned 418, content dump:\n{_request.content}')
+
+        raise Exceptions.FAPIDownForMaintenance
+
+    return _request

FAPI/__init__.py

@@ -0,0 +1 @@
+from . import Queries