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,26 @@ def get_versions_for_playlist(self, playlist_id: int) -> list["Version"]:
         """
         raise NotImplementedError("Subclasses must implement this method.")
 
+    # Removed authenticate_user static method as it is now handled by the module-level factory function below.
+
 
-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}")
+
+
+def authenticate_user(username: str, password: str) -> dict[str, Any]:
+    """Authenticate a user using the configured provider."""
+    provider_type = os.getenv("PRODTRACK_PROVIDER", "shotgrid")
+
+    if provider_type == "shotgrid":
+        from dna.prodtrack_providers.shotgrid_auth import ShotgridAuthenticationProvider
+
+        return ShotgridAuthenticationProvider.authenticate(username, password)
+
     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.
 

backend/src/dna/prodtrack_providers/shotgrid_auth.py

@@ -0,0 +1,59 @@
+"""ShotGrid authentication provider implementation."""
+
+import os
+from typing import Any, Optional
+
+from shotgun_api3 import Shotgun
+
+
+class ShotgridAuthenticationProvider:
+    """Provider for ShotGrid authentication."""
+
+    @staticmethod
+    def authenticate(
+        username: str, password: str, provider_url: Optional[str] = None
+    ) -> dict[str, Any]:
+        """Authenticate a user with ShotGrid and return session token and user info.
+
+        Args:
+            username: User login/username
+            password: User password
+            provider_url: Optional ShotGrid URL. If not provided, uses SHOTGRID_URL env var.
+
+        Returns:
+            Dictionary containing 'token' and 'email'
+
+        Raises:
+            ValueError: If authentication fails
+        """
+        url = provider_url or 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()
+
+            # We need to fetch the user details (email)
+            # We can use the same connection object 'sg' if it supports find/find_one after auth
+            # OR create a new connection with the token.
+            # Using the existing 'sg' instance is more efficient as it's already authenticated/connected (presumably).
+            # However, shotgun_api3 behaviour: 'sg' initialized with login/pass IS valid.
+
+            # Implementation note: We need to find the user by login to get the email.
+            user_data = sg.find_one(
+                "HumanUser", filters=[["login", "is", username]], fields=["email"]
+            )
+
+            if not user_data:
+                raise ValueError(f"User not found: {username}")
+
+            email = user_data.get("email")
+            if not email:
+                raise ValueError("User has no email address configured")
+
+            return {"token": token, "email": email}
+
+        except Exception as e:
+            raise ValueError(f"Authentication failed: {str(e)}")

backend/src/main.py

@@ -1,10 +1,10 @@
 """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, Header, HTTPException
 from fastapi.middleware.cors import CORSMiddleware
+from pydantic import BaseModel
 
 from dna.models import (
     Asset,
@@ -21,9 +21,16 @@
 from dna.models.entity import ENTITY_MODELS, EntityBase
 from dna.prodtrack_providers.prodtrack_provider_base import (
     ProdtrackProviderBase,
+    authenticate_user,
     get_prodtrack_provider,
 )
 
+
+class LoginRequest(BaseModel):
+    username: str
+    password: str
+
+
 # API metadata for Swagger documentation
 API_TITLE = "DNA Backend"
 API_DESCRIPTION = """
@@ -47,6 +54,10 @@
 
 # Define API tags for organizing endpoints
 tags_metadata = [
+    {
+        "name": "Auth",
+        "description": "Authentication endpoints",
+    },
     {
         "name": "Health",
         "description": "Health check and status endpoints",
@@ -126,17 +137,48 @@
 # -----------------------------------------------------------------------------
 
 
-@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 authenticate_user(request.username, request.password)
+    except ValueError as e:
+        raise HTTPException(status_code=401, detail=str(e))
+
+
 # -----------------------------------------------------------------------------
 # Health endpoints
 # -----------------------------------------------------------------------------