Subscriptions get and create handlers to work with resources

backend/moonstream/api.py

@@ -8,6 +8,7 @@ from fastapi import FastAPI, Form
 from fastapi.middleware.cors import CORSMiddleware
 
 from . import data
+from .routes.subscriptions import app as subscriptions_api
 from .routes.users import app as users_api
 from .settings import ORIGINS, bugout_client as bc, MOONSTREAM_APPLICATION_ID
 from .version import MOONSTREAM_VERSION
@@ -36,4 +37,5 @@ async def version_handler() -> data.VersionResponse:
     return data.VersionResponse(version=MOONSTREAM_VERSION)
 
 
+app.mount("/subscriptions", subscriptions_api)
 app.mount("/users", users_api)

backend/moonstream/data.py

@@ -1,6 +1,9 @@
 """
 Pydantic schemas for the Moonstream HTTP API
 """
+import uuid
+from typing import List
+
 from pydantic import BaseModel
 
 
@@ -18,3 +21,16 @@ class VersionResponse(BaseModel):
     """
 
     version: str
+
+
+class SubscriptionRequest(BaseModel):
+    blockchain: str
+
+
+class SubscriptionResponse(BaseModel):
+    user_id: str
+    blockchain: str
+
+
+class SubscriptionsListResponse(BaseModel):
+    subscriptions: List[SubscriptionResponse]

backend/moonstream/routes/subscriptions.py

@@ -0,0 +1,101 @@
+"""
+The Moonstream subscriptions HTTP API
+"""
+import logging
+from typing import Any, Dict
+import uuid
+
+from bugout.data import BugoutResource, BugoutResources
+from bugout.exceptions import BugoutResponseException
+from fastapi import Body, FastAPI, Form, HTTPException, Request, Query
+from fastapi.middleware.cors import CORSMiddleware
+
+from .. import data
+from ..middleware import BroodAuthMiddleware
+from ..settings import (
+    MOONSTREAM_APPLICATION_ID,
+    DOCS_TARGET_PATH,
+    ORIGINS,
+    DOCS_PATHS,
+    bugout_client as bc,
+)
+from ..version import MOONSTREAM_VERSION
+
+logger = logging.getLogger(__name__)
+
+tags_metadata = [
+    {"name": "subscriptions", "description": "Operations with subscriptions."},
+]
+
+app = FastAPI(
+    title=f"Moonstream API.",
+    description="The Bugout blockchain inspector API.",
+    version=MOONSTREAM_VERSION,
+    openapi_tags=tags_metadata,
+    openapi_url="/openapi.json",
+    docs_url=None,
+    redoc_url=f"/{DOCS_TARGET_PATH}",
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=ORIGINS,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+whitelist_paths: Dict[str, str] = {}
+whitelist_paths.update(DOCS_PATHS)
+app.add_middleware(BroodAuthMiddleware, whitelist=whitelist_paths)
+
+
+@app.post("/", tags=["subscriptions"], response_model=data.SubscriptionResponse)
+async def add_subscription_handler(
+    request: Request, subscription_data: data.SubscriptionRequest = Body(...)
+) -> data.SubscriptionResponse:
+    """
+    Add subscription to blockchain stream data for user.
+    """
+    token = request.state.token
+    user = request.state.user
+    resource_data = {"user_id": str(user.id)}
+    resource_data.update(subscription_data)
+    try:
+        resource: BugoutResource = bc.create_resource(
+            token=token,
+            application_id=MOONSTREAM_APPLICATION_ID,
+            resource_data=resource_data,
+        )
+    except BugoutResponseException as e:
+        raise HTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception as e:
+        raise HTTPException(status_code=500)
+    return data.SubscriptionResponse(
+        user_id=resource.resource_data["user_id"],
+        blockchain=resource.resource_data["blockchain"],
+    )
+
+
+@app.get("/", tags=["subscriptions"], response_model=data.SubscriptionsListResponse)
+async def get_subscriptions_handler(request: Request) -> data.SubscriptionsListResponse:
+    """
+    Get user's subscriptions.
+    """
+    token = request.state.token
+    params = {"user_id": str(request.state.user.id)}
+    try:
+        resources: BugoutResources = bc.list_resources(token=token, params=params)
+    except BugoutResponseException as e:
+        raise HTTPException(status_code=e.status_code, detail=e.detail)
+    except Exception as e:
+        raise HTTPException(status_code=500)
+    return data.SubscriptionsListResponse(
+        subscriptions=[
+            data.SubscriptionResponse(
+                user_id=resource.resource_data["user_id"],
+                blockchain=resource.resource_data["blockchain"],
+            )
+            for resource in resources.resources
+        ]
+    )