Add service field to OAuth2Token model

lms/db/_columns.py

@@ -1,7 +1,14 @@
 import sqlalchemy as sa
 
 
-def varchar_enum(enum, max_length=64, nullable=False, unique=False) -> sa.Column:
+def varchar_enum(  # pylint:disable=too-many-arguments
+    enum,
+    default=None,
+    max_length=64,
+    nullable=False,
+    server_default=None,
+    unique=False,
+) -> sa.Column:
     """Return a SA column type to store the python enum.Enum as a varchar in a table."""
     return sa.Column(
         sa.Enum(
@@ -18,6 +25,8 @@ def varchar_enum(enum, max_length=64, nullable=False, unique=False) -> sa.Column
             # Use the string values, not the keys to persist the values
             values_callable=lambda obj: [item.value for item in obj],
         ),
+        default=default,
         nullable=nullable,
+        server_default=server_default,
         unique=unique,
     )

lms/models/oauth2_token.py

@@ -1,14 +1,34 @@
 """Database model for persisting OAuth 2 tokens."""
 
 import datetime
+from enum import Enum, unique
 
 import sqlalchemy as sa
 
-from lms.db import Base
+from lms.db import Base, varchar_enum
 
 __all__ = ["OAuth2Token"]
 
 
+@unique
+class Service(str, Enum):
+    """Enum of the different APIs that OAuth tokens may be used for."""
+
+    LMS = "lms"
+    """
+    The main API of the LMS that a user belongs to.
+
+    This is the API used for resources that are built-in to the LMS.
+    """
+
+    CANVAS_STUDIO = "canvas_studio"
+    """
+    Canvas Studio API.
+
+    See https://tw.instructuremedia.com/api/public/docs/.
+    """
+
+
 class OAuth2Token(Base):
     """
     An OAuth 2.0 access token, refresh token and expiry time.
@@ -75,3 +95,7 @@ class OAuth2Token(Base):
         server_default=sa.func.now(),  # pylint:disable=not-callable
         nullable=False,
     )
+
+    #: The API that this token is used with. In OAuth 2.0 parlance, this
+    #: identifies which kind of resource server the token can be passed to.
+    service = varchar_enum(Service, default="lms", server_default="lms", nullable=False)

tests/unit/lms/models/oauth2_token_test.py

@@ -4,6 +4,7 @@
 import sqlalchemy.exc
 
 from lms.models import OAuth2Token
+from lms.models.oauth2_token import Service
 
 
 class TestOAuth2Token:
@@ -18,6 +19,7 @@ def test_persist_and_retrieve_all_attrs(self, application_instance, db_session):
                 refresh_token="test_refresh_token",
                 expires_in=3600,
                 received_at=now,
+                service=Service.CANVAS_STUDIO,
             )
         )
 
@@ -29,6 +31,7 @@ def test_persist_and_retrieve_all_attrs(self, application_instance, db_session):
         assert token.refresh_token == "test_refresh_token"
         assert token.expires_in == 3600
         assert token.received_at == now
+        assert token.service == Service.CANVAS_STUDIO
 
     @pytest.mark.parametrize("column", ["user_id", "access_token"])
     def test_columns_that_cant_be_None(self, db_session, init_kwargs, column):
@@ -51,29 +54,16 @@ def test_setting_application_instance_sets_fk(
 
         assert token.application_instance_id == application_instance.id
 
-    def test_refresh_token_defaults_to_None(self, db_session, init_kwargs):
+    def test_defaults(self, db_session, init_kwargs):
         token = OAuth2Token(**init_kwargs)
         db_session.add(token)
 
         db_session.flush()
 
         assert token.refresh_token is None
-
-    def test_expires_in_defaults_to_None(self, db_session, init_kwargs):
-        token = OAuth2Token(**init_kwargs)
-        db_session.add(token)
-
-        db_session.flush()
-
         assert token.expires_in is None
-
-    def test_received_at_defaults_to_now(self, db_session, init_kwargs):
-        token = OAuth2Token(**init_kwargs)
-        db_session.add(token)
-
-        db_session.flush()
-
         assert isinstance(token.received_at, datetime.datetime)
+        assert token.service == Service.LMS
 
     @pytest.fixture
     def init_kwargs(self, application_instance):