Adding database support, basic reads

groove/cli.py

@@ -3,12 +3,28 @@ import os
 import typer
 
 from dotenv import load_dotenv
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
 from groove import ondemand
+from groove.db import metadata
 
 
 app = typer.Typer()
 
 
+@app.command()
+def initialize():
+    load_dotenv()
+
+    # todo: abstract this and replace in_memory_db fixture
+    engine = create_engine(f"sqlite:///{os.environ.get('DATABASE_PATH')}", future=True)
+    Session = sessionmaker(bind=engine, future=True)
+    session = Session()
+    metadata.create_all(bind=engine)
+    session.close()
+
+
 @app.command()
 def server(
     host: str = typer.Argument(
@@ -29,6 +45,8 @@ def server(
     """
     load_dotenv()
 
+    ondemand.initialize()
+
     print("Starting Groove On Demand...")
 
     debug = os.getenv('DEBUG', None)

groove/db.py

@@ -0,0 +1,29 @@
+from sqlalchemy import MetaData
+from sqlalchemy import Table, Column, Integer, String, UnicodeText, ForeignKey, PrimaryKeyConstraint
+
+metadata = MetaData()
+
+track = Table(
+    "track",
+    metadata,
+    Column("id", Integer, primary_key=True, autoincrement=True),
+    Column("relpath", UnicodeText, index=True, unique=True),
+)
+
+playlist = Table(
+    "playlist",
+    metadata,
+    Column("id", Integer, primary_key=True, autoincrement=True),
+    Column("name", String),
+    Column("description", UnicodeText),
+    Column("slug", String, index=True, unique=True),
+)
+
+entry = Table(
+    "entry",
+    metadata,
+    Column("track", Integer),
+    Column("playlist_id", Integer, ForeignKey("playlist.id")),
+    Column("track_id", Integer, ForeignKey("track.id")),
+    PrimaryKeyConstraint("playlist_id", "track"),
+)

groove/helper.py

@@ -0,0 +1,49 @@
+import json
+import logging
+
+from bottle import HTTPResponse
+
+from sqlalchemy import bindparam
+
+from groove import db
+
+
+class PlaylistDatabaseHelper:
+    """
+    Convenience class for database interactions.
+    """
+    def __init__(self, connection):
+        self._conn = connection
+
+    @property
+    def conn(self):
+        return self._conn
+
+    def playlist(self, slug: str) -> dict:
+        """
+        Retrieve a playlist and its entries by its slug.
+        """
+        playlist = {}
+
+        query = db.playlist.select(db.playlist.c.slug==bindparam('slug'))
+        logging.debug(f"playlist: '{slug}' requested. Query: {query}")
+        results = self.conn.execute(str(query), {'slug': slug}).fetchone()
+        if not results:
+            return playlist
+
+        playlist = results
+        query = db.entry.select(db.entry.c.playlist_id == bindparam('playlist_id'))
+        logging.debug(f"Retrieving playlist entries. Query: {query}")
+        entries = self.conn.execute(str(query), {'playlist_id': playlist['id']}).fetchall()
+
+        playlist = dict(playlist)
+        playlist['entries'] = [dict(entry) for entry in entries]
+        return playlist
+
+    def json_response(self, playlist: dict, status: int = 200) -> HTTPResponse:
+        """
+        Create an application/json HTTPResponse object out of a playlist and its entries.
+        """
+        response = json.dumps(playlist)
+        logging.debug(response)
+        return HTTPResponse(status=status, content_type='application/json', body=response)

groove/ondemand.py

@@ -1,7 +1,21 @@
-from bottle import Bottle, auth_basic
-from groove.auth import is_authenticated
+import logging
+import os
 
-server = Bottle()
+import bottle
+from bottle import HTTPResponse
+from bottle.ext import sqlite
+
+from groove.auth import is_authenticated
+from groove.helper import PlaylistDatabaseHelper
+
+server = bottle.Bottle()
+
+
+def initialize():
+    """
+    Configure the sqlite database.
+    """
+    server.install(sqlite.Plugin(dbfile=os.environ.get('DATABASE_PATH')))
 
 
 @server.route('/')
@@ -9,7 +23,20 @@ def index():
     return "Groovy."
 
 
-@server.route('/admin')
-@auth_basic(is_authenticated)
-def admin():
+@server.route('/build')
+@bottle.auth_basic(is_authenticated)
+def build():
     return "Authenticated. Groovy."
+
+
+@server.route('/playlist/<slug>')
+def get_playlist(slug, db):
+    """
+    Retrieve a playlist and its entries by a slug.
+    """
+    logging.debug(f"Looking up playlist: {slug}...")
+    pldb = PlaylistDatabaseHelper(connection=db)
+    playlist = pldb.playlist(slug)
+    if not playlist:
+        return HTTPResponse(status=404, body="Not found")
+    return pldb.json_response(playlist)

pyproject.toml

@@ -14,6 +14,8 @@ bottle = "^0.12.23"
 typer = "^0.7.0"
 python-dotenv = "^0.21.0"
 Paste = "^3.5.2"
+bottle-sqlite = "^0.2.0"
+SQLAlchemy = "^1.4.44"
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.2.0"

test/conftest.py

@@ -0,0 +1,56 @@
+import pytest
+
+import groove.db
+
+from sqlalchemy import create_engine, insert
+from sqlalchemy.orm import sessionmaker
+
+
+@pytest.fixture(scope='function')
+def in_memory_db():
+    """
+    An (empty) in-memory SQLite3 database
+    """
+    engine = create_engine('sqlite:///:memory:', future=True)
+    Session = sessionmaker(bind=engine, future=True)
+    session = Session()
+    groove.db.metadata.create_all(bind=engine)
+    yield session
+    session.close()
+
+
+@pytest.fixture(scope='function')
+def db(in_memory_db):
+    """
+    Populate the in-memory sqlite database with fixture data.
+    """
+
+    # create tracks
+    query = insert(groove.db.track)
+    in_memory_db.execute(query, [
+        {'id': 1, 'relpath': '/UNKLE/Psyence Fiction/01 Guns Blazing (Drums of Death, Part 1).flac'},
+        {'id': 2, 'relpath': '/UNKLE/Psyence Fiction/02 UNKLE (Main Title Theme).flac'},
+        {'id': 3, 'relpath': '/UNKLE/Psyence Fiction/03 Bloodstain.flac'}
+    ])
+
+    # create playlists
+    query = insert(groove.db.playlist)
+    in_memory_db.execute(query, [
+        {'id': 1, 'name': 'playlist one', 'description': 'the first one', 'slug': 'playlist-one'},
+        {'id': 2, 'name': 'playlist two', 'description': 'the second one', 'slug': 'playlist-two'},
+        {'id': 3, 'name': 'playlist three', 'description': 'the threerd one', 'slug': 'playlist-three'}
+    ])
+
+    # populate the playlists
+    query = insert(groove.db.entry)
+    in_memory_db.execute(query, [
+        {'playlist_id': '1', 'track': '1', 'track_id': '1'},
+        {'playlist_id': '1', 'track': '2', 'track_id': '2'},
+        {'playlist_id': '1', 'track': '3', 'track_id': '3'},
+
+        {'playlist_id': '2', 'track': '1', 'track_id': '1'},
+
+        {'playlist_id': '3', 'track': '6', 'track_id': '2'},
+        {'playlist_id': '3', 'track': '2', 'track_id': '3'},
+    ])
+    yield in_memory_db

test/test_ondemand.py

@@ -1,31 +1,59 @@
+import pytest
+
 import os
 import sys
 
 import atheris
+import bottle
 from boddle import boddle
 
 from groove import ondemand
 
 
+@pytest.fixture(autouse=True, scope='session')
+def init_db():
+    ondemand.initialize()
+
+
 def test_server():
     with boddle():
-        assert ondemand.index() == 'Groovy.'
+        ondemand.index()
+        assert bottle.response.status_code == 200
 
 
 def test_auth_with_valid_credentials():
     with boddle(auth=(os.environ.get('USERNAME'), os.environ.get('PASSWORD'))):
-        assert ondemand.admin() == 'Authenticated. Groovy.'
+        ondemand.build()
+        assert bottle.response.status_code == 200
 
 
 def test_auth_random_input():
 
     def auth(fuzzed_input):
         with boddle(auth=(fuzzed_input, fuzzed_input)):
-            result = ondemand.admin()
-            assert result.body == 'Access denied'
+            response = ondemand.build()
+            assert response.status_code == 403
 
     atheris.Setup([sys.argv[0], "-atheris_runs=100000"], auth)
     try:
         atheris.Fuzz()
     except SystemExit:
         pass
+
+
+@pytest.mark.parametrize('slug, expected', [
+    ('playlist-one', 200),
+    ('playlist-two', 200),
+    ('playlist-three', 200),
+    ('no such playlist', 404),
+])
+def test_playlist(db, slug, expected):
+    with boddle():
+        response = ondemand.get_playlist(slug, db)
+        assert response.status_code == expected
+
+
+def test_playlist_on_empty_db(in_memory_db):
+    with boddle():
+        response = ondemand.get_playlist('some-slug', in_memory_db)
+        assert response.status_code == 404