feat: Implement Shotgrid authentication with username/password

backend/src/dna/prodtrack_providers/prodtrack_provider_base.py

@@ -51,6 +51,10 @@ def get_user_by_email(self, user_email: str) -> "User":
         """
         raise NotImplementedError("Subclasses must implement this method.")
 
+    def get_user_by_login(self, login: str) -> "User":
+        """Get a user by their login/username."""
+        raise NotImplementedError("Subclasses must implement this method.")
+
     def get_projects_for_user(self, user_email: str) -> list["Project"]:
         """Get projects accessible by a user.
 
@@ -84,12 +88,17 @@ def get_versions_for_playlist(self, playlist_id: int) -> list["Version"]:
         """
         raise NotImplementedError("Subclasses must implement this method.")
 
+    @staticmethod
+    def authenticate_user(username: str, password: str) -> dict[str, Any]:
+        """Authenticate a user and return a session token and user info."""
+        raise NotImplementedError("Subclasses must implement this method.")
+
 
-def get_prodtrack_provider() -> ProdtrackProviderBase:
+def get_prodtrack_provider(session_token: str | None = None) -> ProdtrackProviderBase:
     """Get the production tracking provider."""
     from dna.prodtrack_providers.shotgrid import ShotgridProvider
 
     provider_type = os.getenv("PRODTRACK_PROVIDER", "shotgrid")
     if provider_type == "shotgrid":
-        return ShotgridProvider()
+        return ShotgridProvider(session_token=session_token)
     raise ValueError(f"Unknown production tracking provider: {provider_type}")

backend/src/dna/prodtrack_providers/shotgrid.py

@@ -122,6 +122,7 @@ def __init__(
         url: Optional[str] = None,
         script_name: Optional[str] = None,
         api_key: Optional[str] = None,
+        session_token: Optional[str] = None,
         connect: bool = True,
     ):
         """Initialize the ShotGrid connection.
@@ -130,17 +131,22 @@ def __init__(
             url: ShotGrid server URL. Defaults to SHOTGRID_URL env var.
             script_name: API script name. Defaults to SHOTGRID_SCRIPT_NAME env var.
             api_key: API key for authentication. Defaults to SHOTGRID_API_KEY env var.
+            session_token: Session token for user authentication.
         """
         super().__init__()
 
         self.url = url or os.getenv("SHOTGRID_URL")
         self.script_name = script_name or os.getenv("SHOTGRID_SCRIPT_NAME")
         self.api_key = api_key or os.getenv("SHOTGRID_API_KEY")
+        self.session_token = session_token
 
-        if not all([self.url, self.script_name, self.api_key]):
+        if not self.url:
+            raise ValueError("ShotGrid URL not provided.")
+
+        if not self.session_token and not (self.script_name and self.api_key):
             raise ValueError(
-                "ShotGrid credentials not provided. Set SHOTGRID_URL, "
-                "SHOTGRID_SCRIPT_NAME, and SHOTGRID_API_KEY environment variables."
+                "ShotGrid credentials not provided. Provide either session_token "
+                "or (script_name and api_key)."
             )
 
         self.sg = None
@@ -149,7 +155,11 @@ def __init__(
 
     def _connect(self):
         """Connect to ShotGrid."""
-        self.sg = Shotgun(self.url, self.script_name, self.api_key)
+        if self.session_token:
+            # When using session token, we don't use script credentials
+            self.sg = Shotgun(self.url, session_token=self.session_token)
+        else:
+            self.sg = Shotgun(self.url, self.script_name, self.api_key)
 
     def _convert_sg_entity_to_dna_entity(
         self,
@@ -398,6 +408,35 @@ def get_user_by_email(self, user_email: str) -> User:
             sg_user, entity_mapping, "user", resolve_links=False
         )
 
+    def get_user_by_login(self, login: str) -> User:
+        """Get a user by their login/username.
+
+        Args:
+            login: The login/username of the user
+
+        Returns:
+            User entity
+
+        Raises:
+            ValueError: If user is not found
+        """
+        if not self.sg:
+            raise ValueError("Not connected to ShotGrid")
+
+        sg_user = self.sg.find_one(
+            "HumanUser",
+            filters=[["login", "is", login]],
+            fields=["id", "name", "email", "login"],
+        )
+
+        if not sg_user:
+            raise ValueError(f"User not found: {login}")
+
+        entity_mapping = FIELD_MAPPING["user"]
+        return self._convert_sg_entity_to_dna_entity(
+            sg_user, entity_mapping, "user", resolve_links=False
+        )
+
     def get_projects_for_user(self, user_email: str) -> list[Project]:
         """Get projects accessible by a user.
 
@@ -538,6 +577,42 @@ def get_versions_for_playlist(self, playlist_id: int) -> list[Version]:
         return versions
 
 
+    @staticmethod
+    def authenticate_user(username: str, password: str) -> dict[str, Any]:
+        """Authenticate a user with ShotGrid and return session token and user info.
+
+        Args:
+            username: User login/username
+            password: User password
+
+        Returns:
+            Dictionary containing 'token' and 'email'
+
+        Raises:
+            ValueError: If authentication fails
+        """
+        url = os.getenv("SHOTGRID_URL")
+        if not url:
+            raise ValueError("SHOTGRID_URL not configured")
+
+        try:
+            # Initialize connection to verify credentials and get token
+            sg = Shotgun(url, login=username, password=password)
+            token = sg.get_session_token()
+
+            # Create a provider instance with the new token to fetch user details
+            # This reuses the existing entity mapping logic
+            provider = ShotgridProvider(url=url, session_token=token)
+            user = provider.get_user_by_login(username)
+
+            if not user.email:
+                raise ValueError("User has no email address configured")
+
+            return {"token": token, "email": user.email}
+
+        except Exception as e:
+            raise ValueError(f"Authentication failed: {str(e)}")
+
 def _get_dna_entity_type(sg_entity_type: str) -> str:
     """Get the DNA entity type from the ShotGrid entity type."""
     for entity_type, entity_data in FIELD_MAPPING.items():

backend/src/main.py

@@ -1,10 +1,11 @@
 """FastAPI application entry point."""
 
-from functools import lru_cache
+
 from typing import Annotated, cast
 
-from fastapi import Depends, FastAPI, HTTPException
+from fastapi import Depends, FastAPI, HTTPException, Header
 from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 
 from dna.models import (
     Asset,
@@ -23,6 +24,13 @@
     ProdtrackProviderBase,
     get_prodtrack_provider,
 )
+from dna.prodtrack_providers.shotgrid import ShotgridProvider
+
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
 
 # API metadata for Swagger documentation
 API_TITLE = "DNA Backend"
@@ -47,6 +55,10 @@
 
 # Define API tags for organizing endpoints
 tags_metadata = [
+    {
+        "name": "Auth",
+        "description": "Authentication endpoints",
+    },
     {
         "name": "Health",
         "description": "Health check and status endpoints",
@@ -126,17 +138,47 @@
 # -----------------------------------------------------------------------------
 
 
-@lru_cache
-def get_prodtrack_provider_cached() -> ProdtrackProviderBase:
-    """Get or create the production tracking provider singleton."""
-    return get_prodtrack_provider()
+def get_token_header(authorization: Annotated[str | None, Header()] = None) -> str | None:
+    """Extract token from Authorization header."""
+    if not authorization:
+        return None
+    if authorization.startswith("Bearer "):
+        return authorization.split(" ")[1]
+    return authorization
+
+
+def get_prodtrack_provider_dep(
+    token: Annotated[str | None, Depends(get_token_header)],
+) -> ProdtrackProviderBase:
+    """Get the production tracking provider with user session."""
+    return get_prodtrack_provider(session_token=token)
 
 
 ProdtrackProviderDep = Annotated[
-    ProdtrackProviderBase, Depends(get_prodtrack_provider_cached)
+    ProdtrackProviderBase, Depends(get_prodtrack_provider_dep)
 ]
 
 
+# -----------------------------------------------------------------------------
+# Auth endpoints
+# -----------------------------------------------------------------------------
+
+
+@app.post(
+    "/auth/login",
+    tags=["Auth"],
+    summary="Login to Production Tracking",
+    description="Authenticate with username and password to get a session token.",
+)
+async def login(request: LoginRequest):
+    """Login to ShotGrid."""
+    try:
+        return ShotgridProvider.authenticate_user(request.username, request.password)
+    except ValueError as e:
+        raise HTTPException(status_code=401, detail=str(e))
+
+
+
 # -----------------------------------------------------------------------------
 # Health endpoints
 # -----------------------------------------------------------------------------