From e4d8ae3d88aac184ba7c15c3683630ea8e182e61 Mon Sep 17 00:00:00 2001 From: A Farzat Date: Sun, 12 Oct 2025 14:57:22 +0300 Subject: Rename wsgi directory to api (more appropriate) --- api/__init__.py | 97 ++++++++++++++++++++++ api/utils.py | 14 ++++ tests/integration/flask.py | 172 +++++++++++++++++++++++++++++++++++++++ tests/integration/integration.py | 2 +- tests/integration/wsgi.py | 172 --------------------------------------- wsgi/__init__.py | 97 ---------------------- wsgi/utils.py | 14 ---- 7 files changed, 284 insertions(+), 284 deletions(-) create mode 100644 api/__init__.py create mode 100644 api/utils.py create mode 100644 tests/integration/flask.py delete mode 100644 tests/integration/wsgi.py delete mode 100644 wsgi/__init__.py delete mode 100644 wsgi/utils.py diff --git a/api/__init__.py b/api/__init__.py new file mode 100644 index 0000000..6224e6e --- /dev/null +++ b/api/__init__.py @@ -0,0 +1,97 @@ +from datetime import datetime, UTC +from typing import Any, Dict, List, Tuple + +from flask import Flask, request +from flask_cors import CORS +from pymongo.errors import DuplicateKeyError + +from components.database import subscriptions +from components.subscriptions.main import Subscription +from components.videos import VideoTuple +from components.extractor.extract_sub_info import get_sub_info_from_yt_url +from .utils import vid_dicts_from_tuple_list, sub_info_from_dict + +app = Flask(__name__) +CORS(app) + +@app.route("/vid-from-link/") +def videos_from_link(id: str) -> Tuple[List[Dict[str, Any]], int]: + sub_dict = subscriptions.find_one({"_id": id}) + if sub_dict: + return vid_dicts_from_tuple_list(sub_dict["videos"]), 200 + return [{'error': "Subscription %s not found"%id }], 404 + +@app.route("/sub-info/") +def sub_dict(id: str) -> Tuple[Dict[str, Any], int]: + sub_dict = subscriptions.find_one({"_id": id}) + if sub_dict: + return sub_info_from_dict(sub_dict), 200 + return {'error': "Subscription %s not found"%id }, 404 + +@app.route("/subs-info") +def subs_info() -> List[Dict[str, Any]]: + return [sub_info_from_dict(sub_dict) for sub_dict in subscriptions.find()] + +@app.post("/add-sub/") +def add_sub() -> Tuple[Dict[str, Any], int]: + try: + sub_info = get_sub_info_from_yt_url(request.form["url"]) + time_between_fetches = int(request.form["time_between_fetches"]) + except: + return {'error': 'Invalid data'}, 400 + sub = Subscription( + _id=sub_info["id"], + link=sub_info["link"], + title=sub_info["title"], + time_between_fetches=time_between_fetches, + ) + try: + sub.insert() + return sub_info_from_dict(sub.asdict()), 201 + except DuplicateKeyError: + return {'error': "Subscription %s already exists"%sub_info["id"] }, 409 + +@app.patch("/set-time-between-fetches/") +def set_time_between_fetches(id: str) -> Tuple[Dict[str, Any], int]: + try: + time_between_fetches = int(request.form["time_between_fetches"]) + except: + return {'error': 'Invalid data'}, 400 + result = subscriptions.update_one( + {"_id": id}, + {"$set": {"time_between_fetches": time_between_fetches}} + ) + if result.modified_count: + return { + "_id": id, + "time_between_fetches": time_between_fetches, + }, 200 + return {'error': "Subscription %s not found"%id }, 404 + +@app.delete("/delete-sub/") +def delete_sub(id: str) -> Tuple[Dict[str, Any], int]: + result = subscriptions.delete_one({"_id": id}) + if not result.deleted_count: + return {'error': "Subscription %s not found"%id }, 404 + return { "_id": id, }, 200 + +@app.patch("/set-viewed/") +def set_viewed(id:str) -> Tuple[Dict[str, Any], int]: + viewed_time_str = request.form.get("viewed_time") + if viewed_time_str: + try: + viewed_time = datetime.fromisoformat(viewed_time_str) + except: + return {'error': 'Invalid data'}, 400 + else: + viewed_time = datetime.now(tz=UTC) + result = subscriptions.update_one( + {"_id": id}, + {"$set": {"last_viewed": viewed_time}} + ) + if result.modified_count: + return { + "_id": id, + "last_viewed": viewed_time, + }, 200 + return {'error': "Subscription %s not found"%id }, 404 diff --git a/api/utils.py b/api/utils.py new file mode 100644 index 0000000..2c9e1e5 --- /dev/null +++ b/api/utils.py @@ -0,0 +1,14 @@ +from typing import Any, Dict, List +from components.subscriptions.main import Subscription +from components.subscriptions.typing import SubsDict +from components.videos import VideoTuple + +def vid_dicts_from_tuple_list(tuple_list: List[VideoTuple]) -> List[Dict[str, Any]]: + return [VideoTuple._make(vid)._asdict() for vid in tuple_list] + +def sub_info_from_dict(sub_dict: SubsDict) -> Dict[str, Any]: + return { + **sub_dict, + "videos": len(sub_dict["videos"]), + "new_vids": len(Subscription(**sub_dict).get_new_vids()), + } diff --git a/tests/integration/flask.py b/tests/integration/flask.py new file mode 100644 index 0000000..5597690 --- /dev/null +++ b/tests/integration/flask.py @@ -0,0 +1,172 @@ +from datetime import datetime, UTC +from unittest import TestCase + +from werkzeug.http import parse_date + +from api import app +from components.database import subscriptions + +class TestFlask(TestCase): + def setUp(self) -> None: + app.config['TESTING'] = True + self.client = app.test_client() + subscriptions.delete_many({}) + + def test_add_sub(self) -> None: + # A valid channel - should succeed. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", + 'time_between_fetches': 3123, + }) + self.assertEqual(response.status_code, 201) + # Duplicate - should return 409. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", + 'time_between_fetches': 532, + }) + self.assertEqual(response.status_code, 409) + # A madeup channel - should return 400. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk14Tg--mrJ2A", + 'time_between_fetches': 6824, + }) + self.assertEqual(response.status_code, 400) + # Integer strings are accepted for time_between_fetches. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDMsr9K-rj53DwVRMYO3t5Yr", + 'time_between_fetches': "622", + }) + self.assertEqual(response.status_code, 201) + # An invalid duration - should return 400. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", + 'time_between_fetches': "32.2", + }) + self.assertEqual(response.status_code, 400) + # An invalid subscription - should return 400. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/watch?v=WI4U1SVIO3I", + 'time_between_fetches': 132, + }) + self.assertEqual(response.status_code, 400) + # A valid channel - should succeed. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/@MentalOutlaw/videos", + 'time_between_fetches': 234, + }) + self.assertEqual(response.status_code, 201) + # A valid playlist - should succeed. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/watch?v=kYB8IZa5AuE&list=PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab&index=3", + 'time_between_fetches': 231, + }) + self.assertEqual(response.status_code, 201) + # A duplicate despite the structure of the url being different. + response = self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab", + 'time_between_fetches': 2380, + }) + self.assertEqual(response.status_code, 409) + + def test_subs_info(self) -> None: + # Should be empty at start. + response = self.client.get("/subs-info") + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + self.assertListEqual(response_data, []) + # Fetching any individual subscription now should fail. + response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 404) + # Fetching the videos of an individual subscription should also fail. + response = self.client.get("/vid-from-link/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 404) + # Now add some data to test on... + self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", + 'time_between_fetches': 1, + }) + self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDMsr9K-rj53DwVRMYO3t5Yr", + 'time_between_fetches': 1, + }) + # Now should contain 2 subscriptions. + response = self.client.get("/subs-info") + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + self.assertEqual(len(response_data), 2) + for sub_info in response_data: + self.assertEqual(sub_info["videos"], 0) + # Now fetching an added subscriptions should work. + response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + self.assertEqual(response_data["_id"], "yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + # Same for its videos - though they should still be empty. + response = self.client.get("/vid-from-link/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + self.assertListEqual(response_data, []) + # Subscriptions not added should still not work. + response = self.client.get("/sub-info/yt:channel:7YOGHUfC1Tb6E4pudI9STA") + self.assertEqual(response.status_code, 404) + # Same for for their videos. + response = self.client.get("/vid-from-link/yt:channel:7YOGHUfC1Tb6E4pudI9STA") + self.assertEqual(response.status_code, 404) + + def test_modify_subs(self) -> None: + # Confirm all methods do not work before the item exists. + response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'time_between_fetches': 313, + }) + self.assertEqual(response.status_code, 404) + response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'viewed_time': datetime.now(tz=UTC).isoformat(), + }) + self.assertEqual(response.status_code, 404) + response = self.client.delete("/delete-sub/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 404) + # Now add some data to test on... + self.client.post("/add-sub/", data={ + 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", + 'time_between_fetches': 1, + }) + # Test set-time-between-fetches. + response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'time_between_fetches': 31323, + }) + self.assertEqual(response.status_code, 200) + # Should not work if an integer is not provided. + response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'time_between_fetches': 31.3, + }) + self.assertEqual(response.status_code, 400) + # Test set-viewed. + update_time_to = datetime.now(tz=UTC) + response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'viewed_time': update_time_to.isoformat(), + }) + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + self.assertEqual(update_time_to.replace(microsecond=0), parse_date(response_data["last_viewed"])) + # Should not work with an invalid datetime string. + response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ + 'viewed_time': "Monday 11th of February 2023", + }) + self.assertEqual(response.status_code, 400) + # Should work fine without a viewed_time set though... + time_before_updating = datetime.now(tz=UTC) + response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 200) + response_data = response.get_json() + last_viewed = parse_date(response_data["last_viewed"]) + assert last_viewed + self.assertLessEqual(time_before_updating.replace(microsecond=0), last_viewed) + # Delete the subscription. + response = self.client.delete("/delete-sub/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 200) + # Make sure the item no longer exists. + response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") + self.assertEqual(response.status_code, 404) + + def tearDown(self) -> None: + subscriptions.delete_many({}) diff --git a/tests/integration/integration.py b/tests/integration/integration.py index e3b103c..2577ea0 100644 --- a/tests/integration/integration.py +++ b/tests/integration/integration.py @@ -1,10 +1,10 @@ from os import getenv from unittest import TestCase +from api import app from components.database import subscriptions from data_analyser.utils import analyse_collection from data_collector.utils import collect_data -from wsgi import app class TestIntegration(TestCase): def setUp(self) -> None: diff --git a/tests/integration/wsgi.py b/tests/integration/wsgi.py deleted file mode 100644 index 8124abe..0000000 --- a/tests/integration/wsgi.py +++ /dev/null @@ -1,172 +0,0 @@ -from datetime import datetime, UTC -from unittest import TestCase - -from werkzeug.http import parse_date - -from components.database import subscriptions -from wsgi import app - -class TestFlask(TestCase): - def setUp(self) -> None: - app.config['TESTING'] = True - self.client = app.test_client() - subscriptions.delete_many({}) - - def test_add_sub(self) -> None: - # A valid channel - should succeed. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", - 'time_between_fetches': 3123, - }) - self.assertEqual(response.status_code, 201) - # Duplicate - should return 409. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", - 'time_between_fetches': 532, - }) - self.assertEqual(response.status_code, 409) - # A madeup channel - should return 400. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk14Tg--mrJ2A", - 'time_between_fetches': 6824, - }) - self.assertEqual(response.status_code, 400) - # Integer strings are accepted for time_between_fetches. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDMsr9K-rj53DwVRMYO3t5Yr", - 'time_between_fetches': "622", - }) - self.assertEqual(response.status_code, 201) - # An invalid duration - should return 400. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", - 'time_between_fetches': "32.2", - }) - self.assertEqual(response.status_code, 400) - # An invalid subscription - should return 400. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/watch?v=WI4U1SVIO3I", - 'time_between_fetches': 132, - }) - self.assertEqual(response.status_code, 400) - # A valid channel - should succeed. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/@MentalOutlaw/videos", - 'time_between_fetches': 234, - }) - self.assertEqual(response.status_code, 201) - # A valid playlist - should succeed. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/watch?v=kYB8IZa5AuE&list=PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab&index=3", - 'time_between_fetches': 231, - }) - self.assertEqual(response.status_code, 201) - # A duplicate despite the structure of the url being different. - response = self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDPD3MizzM2xVFitgF8hE_ab", - 'time_between_fetches': 2380, - }) - self.assertEqual(response.status_code, 409) - - def test_subs_info(self) -> None: - # Should be empty at start. - response = self.client.get("/subs-info") - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - self.assertListEqual(response_data, []) - # Fetching any individual subscription now should fail. - response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 404) - # Fetching the videos of an individual subscription should also fail. - response = self.client.get("/vid-from-link/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 404) - # Now add some data to test on... - self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", - 'time_between_fetches': 1, - }) - self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/playlist?list=PLZHQObOWTQDMsr9K-rj53DwVRMYO3t5Yr", - 'time_between_fetches': 1, - }) - # Now should contain 2 subscriptions. - response = self.client.get("/subs-info") - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - self.assertEqual(len(response_data), 2) - for sub_info in response_data: - self.assertEqual(sub_info["videos"], 0) - # Now fetching an added subscriptions should work. - response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - self.assertEqual(response_data["_id"], "yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - # Same for its videos - though they should still be empty. - response = self.client.get("/vid-from-link/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - self.assertListEqual(response_data, []) - # Subscriptions not added should still not work. - response = self.client.get("/sub-info/yt:channel:7YOGHUfC1Tb6E4pudI9STA") - self.assertEqual(response.status_code, 404) - # Same for for their videos. - response = self.client.get("/vid-from-link/yt:channel:7YOGHUfC1Tb6E4pudI9STA") - self.assertEqual(response.status_code, 404) - - def test_modify_subs(self) -> None: - # Confirm all methods do not work before the item exists. - response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'time_between_fetches': 313, - }) - self.assertEqual(response.status_code, 404) - response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'viewed_time': datetime.now(tz=UTC).isoformat(), - }) - self.assertEqual(response.status_code, 404) - response = self.client.delete("/delete-sub/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 404) - # Now add some data to test on... - self.client.post("/add-sub/", data={ - 'url': "https://www.youtube.com/channel/UCBa659QWEk1AI4Tg--mrJ2A", - 'time_between_fetches': 1, - }) - # Test set-time-between-fetches. - response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'time_between_fetches': 31323, - }) - self.assertEqual(response.status_code, 200) - # Should not work if an integer is not provided. - response = self.client.patch("/set-time-between-fetches/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'time_between_fetches': 31.3, - }) - self.assertEqual(response.status_code, 400) - # Test set-viewed. - update_time_to = datetime.now(tz=UTC) - response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'viewed_time': update_time_to.isoformat(), - }) - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - self.assertEqual(update_time_to.replace(microsecond=0), parse_date(response_data["last_viewed"])) - # Should not work with an invalid datetime string. - response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A", data={ - 'viewed_time': "Monday 11th of February 2023", - }) - self.assertEqual(response.status_code, 400) - # Should work fine without a viewed_time set though... - time_before_updating = datetime.now(tz=UTC) - response = self.client.patch("/set-viewed/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 200) - response_data = response.get_json() - last_viewed = parse_date(response_data["last_viewed"]) - assert last_viewed - self.assertLessEqual(time_before_updating.replace(microsecond=0), last_viewed) - # Delete the subscription. - response = self.client.delete("/delete-sub/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 200) - # Make sure the item no longer exists. - response = self.client.get("/sub-info/yt:channel:Ba659QWEk1AI4Tg--mrJ2A") - self.assertEqual(response.status_code, 404) - - def tearDown(self) -> None: - subscriptions.delete_many({}) diff --git a/wsgi/__init__.py b/wsgi/__init__.py deleted file mode 100644 index 6224e6e..0000000 --- a/wsgi/__init__.py +++ /dev/null @@ -1,97 +0,0 @@ -from datetime import datetime, UTC -from typing import Any, Dict, List, Tuple - -from flask import Flask, request -from flask_cors import CORS -from pymongo.errors import DuplicateKeyError - -from components.database import subscriptions -from components.subscriptions.main import Subscription -from components.videos import VideoTuple -from components.extractor.extract_sub_info import get_sub_info_from_yt_url -from .utils import vid_dicts_from_tuple_list, sub_info_from_dict - -app = Flask(__name__) -CORS(app) - -@app.route("/vid-from-link/") -def videos_from_link(id: str) -> Tuple[List[Dict[str, Any]], int]: - sub_dict = subscriptions.find_one({"_id": id}) - if sub_dict: - return vid_dicts_from_tuple_list(sub_dict["videos"]), 200 - return [{'error': "Subscription %s not found"%id }], 404 - -@app.route("/sub-info/") -def sub_dict(id: str) -> Tuple[Dict[str, Any], int]: - sub_dict = subscriptions.find_one({"_id": id}) - if sub_dict: - return sub_info_from_dict(sub_dict), 200 - return {'error': "Subscription %s not found"%id }, 404 - -@app.route("/subs-info") -def subs_info() -> List[Dict[str, Any]]: - return [sub_info_from_dict(sub_dict) for sub_dict in subscriptions.find()] - -@app.post("/add-sub/") -def add_sub() -> Tuple[Dict[str, Any], int]: - try: - sub_info = get_sub_info_from_yt_url(request.form["url"]) - time_between_fetches = int(request.form["time_between_fetches"]) - except: - return {'error': 'Invalid data'}, 400 - sub = Subscription( - _id=sub_info["id"], - link=sub_info["link"], - title=sub_info["title"], - time_between_fetches=time_between_fetches, - ) - try: - sub.insert() - return sub_info_from_dict(sub.asdict()), 201 - except DuplicateKeyError: - return {'error': "Subscription %s already exists"%sub_info["id"] }, 409 - -@app.patch("/set-time-between-fetches/") -def set_time_between_fetches(id: str) -> Tuple[Dict[str, Any], int]: - try: - time_between_fetches = int(request.form["time_between_fetches"]) - except: - return {'error': 'Invalid data'}, 400 - result = subscriptions.update_one( - {"_id": id}, - {"$set": {"time_between_fetches": time_between_fetches}} - ) - if result.modified_count: - return { - "_id": id, - "time_between_fetches": time_between_fetches, - }, 200 - return {'error': "Subscription %s not found"%id }, 404 - -@app.delete("/delete-sub/") -def delete_sub(id: str) -> Tuple[Dict[str, Any], int]: - result = subscriptions.delete_one({"_id": id}) - if not result.deleted_count: - return {'error': "Subscription %s not found"%id }, 404 - return { "_id": id, }, 200 - -@app.patch("/set-viewed/") -def set_viewed(id:str) -> Tuple[Dict[str, Any], int]: - viewed_time_str = request.form.get("viewed_time") - if viewed_time_str: - try: - viewed_time = datetime.fromisoformat(viewed_time_str) - except: - return {'error': 'Invalid data'}, 400 - else: - viewed_time = datetime.now(tz=UTC) - result = subscriptions.update_one( - {"_id": id}, - {"$set": {"last_viewed": viewed_time}} - ) - if result.modified_count: - return { - "_id": id, - "last_viewed": viewed_time, - }, 200 - return {'error': "Subscription %s not found"%id }, 404 diff --git a/wsgi/utils.py b/wsgi/utils.py deleted file mode 100644 index 2c9e1e5..0000000 --- a/wsgi/utils.py +++ /dev/null @@ -1,14 +0,0 @@ -from typing import Any, Dict, List -from components.subscriptions.main import Subscription -from components.subscriptions.typing import SubsDict -from components.videos import VideoTuple - -def vid_dicts_from_tuple_list(tuple_list: List[VideoTuple]) -> List[Dict[str, Any]]: - return [VideoTuple._make(vid)._asdict() for vid in tuple_list] - -def sub_info_from_dict(sub_dict: SubsDict) -> Dict[str, Any]: - return { - **sub_dict, - "videos": len(sub_dict["videos"]), - "new_vids": len(Subscription(**sub_dict).get_new_vids()), - } -- cgit v1.2.3-70-g09d2