diff options
| -rw-r--r-- | cli-project/src/solo_tool_cli.py | 4 | ||||
| -rw-r--r-- | solo-tool-project/src/solo_tool/session_manager.py | 117 | ||||
| -rw-r--r-- | solo-tool-project/test/session_manager_unittest.py | 10 | ||||
| -rw-r--r-- | web-project/src/solo_tool_web.py | 4 |
4 files changed, 74 insertions, 61 deletions
diff --git a/cli-project/src/solo_tool_cli.py b/cli-project/src/solo_tool_cli.py index 34d88dd..89879fc 100644 --- a/cli-project/src/solo_tool_cli.py +++ b/cli-project/src/solo_tool_cli.py @@ -3,7 +3,7 @@ import time from solo_tool import SoloTool from solo_tool.midi_controller_launchpad_mini import MidiController -from solo_tool.session_manager import getSessionManager +from solo_tool.session_manager import SessionManager def main(): args = sys.argv[1:] @@ -11,7 +11,7 @@ def main(): print("Usage: solo_tool_cli <path_to_sessions> <session_id>") sys.exit(1) - sessionManager = getSessionManager(args[0]) + sessionManager = SessionManager(args[0]) soloTool = sessionManager.loadSession(args[1]) midiController = MidiController(soloTool) diff --git a/solo-tool-project/src/solo_tool/session_manager.py b/solo-tool-project/src/solo_tool/session_manager.py index 2e99937..38dc827 100644 --- a/solo-tool-project/src/solo_tool/session_manager.py +++ b/solo-tool-project/src/solo_tool/session_manager.py @@ -1,96 +1,115 @@ -from typing import BinaryIO, Protocol +from typing import Protocol from abc import abstractmethod +from . import SoloTool + +from pathlib import Path from glob import glob import json -from pathlib import Path -from . import SoloTool import requests -class SessionManager(Protocol): - @abstractmethod - def getSessions(self) -> list[str]: - raise NotImplementedError +class SessionManager(): + def __init__(self, sessionPath: str): + self._sessionPath = sessionPath - @abstractmethod - def loadSession(self, key: str, player=None) -> SoloTool: - raise NotImplementedError + from re import search + match = search(r"^([a-z0-9]+://)", sessionPath) + if not match or match.group(0) == "file://": + self._backend = _FileSystemBackend(sessionPath) + elif match.group(0) in ["http://", "https://"]: + self._backend = _FileBrowserBackend(sessionPath) + else: + raise ValueError(f"Unsupported session path: {sessionPath}") - @abstractmethod - def saveSession(self, soloTool: SoloTool, key: str) -> None: - raise NotImplementedError + def getSessions(self) -> list[str]: + return self._backend.listIds() - @staticmethod - def _dict2st(session: dict, player) -> SoloTool: - st = SoloTool(player=player) + def loadSession(self, id: str, player=None) -> SoloTool: + session = self._backend.read(id) + st = SoloTool(player=player) for i, entry in enumerate(session): songPath = entry["path"] keyPoints = entry["key_points"] st.addSong(songPath) st._keyPoints[i] = keyPoints - + return st - @staticmethod - def _st2dict(soloTool: SoloTool) -> dict: + def saveSession(self, soloTool: SoloTool, id: str) -> None: session = [] + for i, song in enumerate(soloTool.songs): entry = { "path": song, "key_points" : soloTool._keyPoints[i] } session.append(entry) - return session -class _FileSystemSessionManager(SessionManager): + self._backend.write(session, id) + +class _Backend(Protocol): + @abstractmethod + def listIds(self) -> list[str]: + raise NotImplementedError + + @abstractmethod + def read(self, id: str) -> dict: + raise NotImplementedError + + @abstractmethod + def write(self, session: dict, id: str) -> None: + raise NotImplementedError + +class _FileSystemBackend(_Backend): def __init__(self, sessionPath: str): self._sessionPath = Path(sessionPath) - def getSessions(self) -> list[str]: + def listIds(self) -> list[str]: return [Path(f).stem for f in glob(f"{self._sessionPath}/*.json")] - def loadSession(self, key: str, player=None) -> SoloTool: - with open(self._sessionPath / f"{key}.json", "r") as f: + def read(self, id: str) -> dict: + with open(self._sessionPath / f"{id}.json", "r") as f: session = json.load(f) - return SessionManager._dict2st(session, player) + return session - def saveSession(self, soloTool: SoloTool, key: str) -> None: - session = SessionManager._st2dict(soloTool) - with open(self._sessionPath / f"{key}.json", "w") as f: + def write(self, session: dict, id: str) -> None: + with open(self._sessionPath / f"{id}.json", "w") as f: json.dump(session, f) -class _FileBrowserSessionManager(SessionManager): - def __init__(self, sessionUrl: str): - self._baseUrl = "https://files.0xf7.com" +class _FileBrowserBackend(_Backend): + def __init__(self, serverUrl: str): + self._baseUrl = serverUrl self._username = "solo-tool" self._password = "mwC0ML8vLpJLPCLHKuxkiOxtIaE" self._apiKey = self._getApiKey() - def getSessions(self) -> list[str]: - url = f"{self._baseUrl}/api/resources/sessions" - response = requests.get(url, headers={"X-Auth":self._apiKey}) - response.raise_for_status() + def listIds(self) -> list[str]: + url = f"{self._baseUrl}/api/resources" + response = self._request("GET", url) return [item["name"][0:-5] for item in response.json()["items"] if item["extension"] == ".json"] - def loadSession(self, key: str, player=None) -> SoloTool: - url = f"{self._baseUrl}/api/raw/sessions/{key}.json" - response = requests.get(url, headers={"X-Auth":self._apiKey}) - response.raise_for_status() - return SessionManager._dict2st(json.loads(response.content), player=player) + def read(self, id: str) -> dict: + url = f"{self._baseUrl}/api/raw/{id}.json" + response = self._request("GET", url) + return json.loads(response.content) - def saveSession(self, soloTool: SoloTool, key: str) -> None: - pass + def write(self, session: dict, id: str) -> None: + url = f"{self._baseUrl}/api/resources/{id}.json" + self._request("PUT", url, json=session) def _getApiKey(self) -> str: response = requests.post(f"{self._baseUrl}/api/login", json={"username":self._username, "password":self._password}) return response.content -def getSessionManager(sessionPath: str) -> SessionManager: - from re import search - match = search(r"^([a-z0-9]+://)", sessionPath) - if not match or match.group(0) == "file://": - return _FileSystemSessionManager(sessionPath) - elif match.group(0) in ["http://", "https://"]: - return _FileBrowserSessionManager(sessionPath) + def _request(self, verb: str, url: str, **kwargs): + headers = {"X-Auth" : self._apiKey} + response = requests.request(verb, url, headers=headers, **kwargs) + if response.status_code == requests.codes.UNAUTHORIZED: + # if unauthorized, the key might have expired + self._apiKey = self._getApiKey() + headers["X-Auth"] = self._apiKey + response = requests.request(verb, url, headers=headers, **kwargs) + response.raise_for_status() + return response diff --git a/solo-tool-project/test/session_manager_unittest.py b/solo-tool-project/test/session_manager_unittest.py index 690bec0..bf0d8d9 100644 --- a/solo-tool-project/test/session_manager_unittest.py +++ b/solo-tool-project/test/session_manager_unittest.py @@ -2,7 +2,7 @@ import pytest from json import loads import os -from solo_tool.session_manager import getSessionManager, _FileSystemSessionManager, _FileBrowserSessionManager +from solo_tool.session_manager import SessionManager from fixtures import soloTool, mockPlayer, testSongs, sessionPath @pytest.fixture @@ -24,7 +24,7 @@ def testSessionFile(sessionPath, testSongs): @pytest.fixture def sessionManager(sessionPath): - return getSessionManager(str(sessionPath)) + return SessionManager(str(sessionPath)) def test_loadSession(sessionManager, mockPlayer, testSessionFile): sessions = sessionManager.getSessions() @@ -64,9 +64,3 @@ def test_loadAndSaveEmptySession(sessionManager, sessionPath, soloTool, tmp_path assert reloadedTool.songs == [] -def test_sessionManagerFactory(): - assert type(getSessionManager("/some_absolute_dir")) is _FileSystemSessionManager - assert type(getSessionManager("file:///some_dir_with_protocol")) is _FileSystemSessionManager - assert type(getSessionManager("some_relative_dir")) is _FileSystemSessionManager - assert type(getSessionManager("http://some_server")) is _FileBrowserSessionManager - assert type(getSessionManager("https://some_secure_server")) is _FileBrowserSessionManager diff --git a/web-project/src/solo_tool_web.py b/web-project/src/solo_tool_web.py index 7c8a48e..ea3de5b 100644 --- a/web-project/src/solo_tool_web.py +++ b/web-project/src/solo_tool_web.py @@ -7,7 +7,7 @@ from starlette.formparsers import MultiPartParser import click from solo_tool import SoloTool -from solo_tool.session_manager import getSessionManager +from solo_tool.session_manager import SessionManager from solo_tool import handlers def fileName(path: str) -> str: @@ -114,7 +114,7 @@ def landingPage(): def start(port, refresh, reload, session_path): global sessionManager - sessionManager = getSessionManager(session_path) + sessionManager = SessionManager(session_path) for key in sessionManager.getSessions(): songTool = sessionManager.loadSession(key) |