Feature/astronomy

app/config.py.example

@@ -28,6 +28,8 @@ PICTURE_EXTENSION = '.png'
 AVATAR_SIZE = (120, 120)
 
 # API-KEYS
+# Get a free API KEY for Astronomy feature @ www.weatherapi.com/signup.aspx
+ASTRONOMY_API_KEY = os.getenv('ASTRONOMY_API_KEY')
 WEATHER_API_KEY = os.getenv('WEATHER_API_KEY')
 
 # EXPORT

app/internal/astronomy.py

@@ -0,0 +1,70 @@
+import datetime
+import functools
+import httpx
+from typing import Dict
+
+from app import config
+
+
+# This feature requires an API KEY - get yours free @ www.weatherapi.com
+
+ASTRONOMY_URL = "http://api.weatherapi.com/v1/astronomy.json"
+NO_API_RESPONSE = "No response from server"
+
+
+@functools.lru_cache(maxsize=128, typed=False)
+async def get_data_from_api(formatted_date: str, location: str)\
+        -> Dict[str, int]:
+    """ get the relevant astronomical data by calling the "weather api" API.
+    Args:
+        formatted_date (date) - relevant date.
+        location (str) - location name.
+    Returns:
+        response_json (json dict) including:
+        relevant part (data / error) of the JSON returned by the API.
+        Success (bool)
+        ErrorDescription (str) - error message.
+    """
+    input_query_string = {'key': config.ASTRONOMY_API_KEY, 'q': location,
+                          'dt': formatted_date}
+    output = {}
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.get(ASTRONOMY_URL,
+                                        params=input_query_string)
+    except httpx.HTTPError:
+        output["Success"] = False
+        output["ErrorDescription"] = NO_API_RESPONSE
+        return output
+    if response.status_code != httpx.codes.OK:
+        output["Success"] = False
+        output["ErrorDescription"] = NO_API_RESPONSE
+        return output
+    output["Success"] = True
+    try:
+        output.update(response.json()['location'])
+        return output
+    except KeyError:
+        output["Success"] = False
+        output["ErrorDescription"] = response.json()['error']['message']
+        return output
+
+
+async def get_astronomical_data(requested_date: datetime.datetime,
+                                location: str) -> Dict[str, int]:
+    """ get astronomical data (Sun & Moon) for date & location -
+        main function.
+    Args:
+        requested_date (date) - date requested for astronomical data.
+        location (str) - location name.
+    Returns: dictionary with the following entries:
+        Status - success / failure.
+        ErrorDescription - error description (relevant only in case of error).
+        location - relevant location values(relevant only in case of success).
+            name, region, country, lat, lon etc.
+        astronomy - relevant astronomy values, all time in local time -
+            (relevant only in case of success):
+            sunrise, sunset, moonrise, moonset, moon_phase, moon_illumination.
+    """
+    formatted_date = requested_date.strftime('%Y-%m-%d')
+    return await get_data_from_api(formatted_date, location)

tests/test_astronomy.py

@@ -0,0 +1,70 @@
+import datetime
+import httpx
+import pytest
+import requests
+import responses
+import respx
+
+
+from app.internal.astronomy import get_astronomical_data
+from app.internal.astronomy import ASTRONOMY_URL
+
+RESPONSE_FROM_MOCK = {"location": {
+        "name": "Tel Aviv-Yafo",
+        "region": "Tel Aviv",
+        "country": "Israel",
+        "lat": 32.07,
+        "lon": 34.76,
+        "tz_id": "Asia/Jerusalem",
+        "localtime_epoch": 1611399607,
+        "localtime": "2021-01-23 13:00"
+    },
+    "astronomy": {
+        "astro": {
+            "sunrise": "05:25 AM",
+            "sunset": "06:03 PM",
+            "moonrise": "01:56 PM",
+            "moonset": "03:04 AM",
+            "moon_phase": "Waxing Gibbous",
+            "moon_illumination": "79"
+        }
+    }
+}
+ERROR_RESPONSE_FROM_MOCK = {"error": {"message": "Error Text"}}
+
+
+@pytest.mark.asyncio
+async def test_get_astronomical_data(httpx_mock):
+    requested_date = datetime.datetime(day=4, month=4, year=2020)
+    httpx_mock.add_response(method="GET", json=RESPONSE_FROM_MOCK)
+    output = await get_astronomical_data(requested_date, "tel aviv")
+    assert output['Success']
+
+
+@respx.mock
+@pytest.mark.asyncio
+async def test_astronomical_data_error_from_api():
+    requested_date = datetime.datetime(day=4, month=4, year=2021)
+    route = respx.get(ASTRONOMY_URL)
+    route.return_value = httpx.Response(200, json=ERROR_RESPONSE_FROM_MOCK)
+    output = await get_astronomical_data(requested_date, "123")
+    assert not output['Success']
+
+
+@respx.mock
+@pytest.mark.asyncio
+async def test_astronomical_exception_from_api(httpx_mock):
+    requested_date = datetime.datetime.now() + datetime.timedelta(days=3)
+    respx.get(ASTRONOMY_URL).mock(return_value=httpx.Response(500))
+    output = await get_astronomical_data(requested_date, "456")
+    assert not output['Success']
+
+
+@responses.activate
+@pytest.mark.asyncio
+async def test_astronomical_no_response_from_api():
+    requested_date = datetime.datetime(day=11, month=1, year=2020)
+    responses.add(responses.GET, ASTRONOMY_URL, status=500)
+    requests.get(ASTRONOMY_URL)
+    output = await get_astronomical_data(requested_date, "789")
+    assert not output['Success']