Feature/workflow pretty

.gitignore

@@ -1,4 +1,5 @@
 __pycache__
+.venv
 venv
 local/
 .env

backend/.local.env

@@ -1,8 +1,21 @@
+# --- General Config ---
 FRONTEND_URL="http://localhost:4200"
 ENVIRONMENT="development"
 LOG_LEVEL="INFO"
 PROJECT_ID="<YOUR_GCP_PROJECT_ID>"
+GOOGLE_CLOUD_PROJECT="<YOUR_GCP_PROJECT_ID>"
 GENMEDIA_BUCKET="${PROJECT_ID}-genmedia"
 SIGNING_SA_EMAIL="<YOUR_GCP_SA_EMAIL>"
 GOOGLE_TOKEN_AUDIENCE="<YOUR_GOOGLE_TOKEN>"
 IDENTITY_PLATFORM_ALLOWED_ORGS=""
+
+# --- Database Configuration (Cloud SQL) ---
+INSTANCE_CONNECTION_NAME=your_gcp_project_ID:us-central1:db_instance_name 
+DB_USER=studio_user
+DB_PASS="<YOUR_PASSWORD>"
+DB_NAME=creative_studio
+USE_CLOUD_SQL_AUTH_PROXY=false
+ADMIN_USER_EMAIL="<YOUR_ADMIN_USER>"
+
+# --- For Workflows ---
+SERVICE_ACCOUNT_EMAIL="<YOUR_GCP_SA_EMAIL>"

backend/alembic/env.py

@@ -28,6 +28,7 @@
 from src.common.schema.media_item_model import MediaItem
 from src.media_templates.schema.media_template_model import MediaTemplate
 from src.source_assets.schema.source_asset_model import SourceAsset
+from src.workflows.schema.workflow_model import Workflow
 
 # this is the Alembic Config object, which provides
 # access to the values within the .ini file in use.

backend/alembic/versions/488846b86e90_remove_workspace_id_from_workflows.py

@@ -0,0 +1,46 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""remove workspace_id from workflows
+
+Revision ID: 488846b86e90
+Revises: 73b7bec44615
+Create Date: 2025-12-18 16:47:16.267554
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = '488846b86e90'
+down_revision: Union[str, None] = '73b7bec44615'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(op.f('workflows_workspace_id_fkey'), 'workflows', type_='foreignkey')
+    op.drop_column('workflows', 'workspace_id')
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('workflows', sa.Column('workspace_id', sa.INTEGER(), autoincrement=False, nullable=False))
+    op.create_foreign_key(op.f('workflows_workspace_id_fkey'), 'workflows', 'workspaces', ['workspace_id'], ['id'])
+    # ### end Alembic commands ###

backend/alembic/versions/73b7bec44615_create_workflows_table.py

@@ -0,0 +1,56 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""create workflows table
+
+Revision ID: 73b7bec44615
+Revises: 6591e10bbab7
+Create Date: 2025-12-17 12:12:37.380535
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision: str = '73b7bec44615'
+down_revision: Union[str, None] = '6591e10bbab7'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('workflows',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('workspace_id', sa.Integer(), nullable=False),
+    sa.Column('user_id', sa.Integer(), nullable=False),
+    sa.Column('name', sa.String(), nullable=False),
+    sa.Column('description', sa.String(), nullable=True),
+    sa.Column('steps', postgresql.JSONB(astext_type=sa.Text()), nullable=False),
+    sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), server_default=sa.text('now()'), nullable=False),
+    sa.ForeignKeyConstraint(['user_id'], ['users.id'], ),
+    sa.ForeignKeyConstraint(['workspace_id'], ['workspaces.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('workflows')
+    # ### end Alembic commands ###

backend/main.py

@@ -45,6 +45,11 @@
 )
 from src.users.user_controller import router as user_router
 from src.videos.veo_controller import router as video_router
+
+from src.workflows.workflow_controller import router as workflow_router
+from src.workflows_executor.workflows_executor_controller import (
+    router as workflows_executor_router,
+)
 from src.workspaces.workspace_controller import router as workspace_router
 
 # Get a logger instance for use in this file. It will inherit the root setup.
@@ -117,7 +122,6 @@ async def lifespan(app: FastAPI):
     app.state.executor.shutdown(wait=True)
     # Your shutdown logic here, e.g., closing database connections
 
-
 app = FastAPI(
     lifespan=lifespan,
     title="Creative Studio API",
@@ -168,3 +172,5 @@ def version():
 app.include_router(source_asset_router)
 app.include_router(workspace_router)
 app.include_router(brand_guideline_router)
+app.include_router(workflow_router)
+app.include_router(workflows_executor_router)

backend/pyproject.toml

@@ -32,6 +32,9 @@ dependencies = [
     "asyncpg>=0.29.0",
     "cloud-sql-python-connector[asyncpg]>=1.0.0",
     "alembic>=1.13.0",
+    "pyyaml>=6.0.3",
+    "google-cloud-workflows>=1.19.0",
+    "google-api-python-client>=2.187.0",
 ]
 
 [project.optional-dependencies]
@@ -70,3 +73,8 @@ addopts = "--cov --cov-report=lcov:lcov.info --cov-report=term"
 
 [tool.isort]
 profile = "black"
+
+[[tool.uv.index]]
+name = "pypi"
+url = "https://pypi.org/simple"
+default = true

backend/src/audios/audio_service.py

@@ -354,7 +354,7 @@ async def generate_single_sample(index: int) -> Optional[str]:
                 # We take the first one because we requested sample_count=1
                 prediction = response.predictions[0]
                 # prediction is a MapComposite, which behaves like a dict
-                audio_b64 = prediction.get("bytesBase64Encoded")
+                audio_b64 = prediction.get("bytesBase64Encoded")  # type: ignore
 
                 if not audio_b64:
                     return None

backend/src/brand_guidelines/brand_guideline_service.py

@@ -22,7 +22,6 @@
 import sys
 import uuid
 from concurrent.futures import (
-    ProcessPoolExecutor,
     ThreadPoolExecutor,
     as_completed,
 )
@@ -73,7 +72,7 @@ def _process_brand_guideline_in_background(
     workspace_id: Optional[int],
 ):
     """
-    This is the long-running worker task that runs in a separate process.
+    This is the long-running worker task that runs in a separate thread.
     It handles PDF splitting, uploading, AI extraction, and database updates.
     """
     import asyncio

backend/src/common/base_dto.py

@@ -28,7 +28,7 @@ class MimeTypeEnum(str, Enum):
 
 
 class GenerationModelEnum(str, Enum):
-    """Enum representing the available Imagen generation models."""
+    """Enum representing the available generation models."""
 
     # Image-Specific Models
     IMAGEN_4_001 = "imagen-4.0-generate-001"
@@ -42,10 +42,12 @@ class GenerationModelEnum(str, Enum):
     IMAGEGEN_006 = "imagegeneration@006"
     IMAGEGEN_005 = "imagegeneration@005"
     IMAGEGEN_002 = "imagegeneration@002"
-    GEMINI_2_5_FLASH_IMAGE_PREVIEW = "gemini-2.5-flash-image-preview"
-    GEMINI_3_PRO_IMAGE_PREVIEW = "gemini-3-pro-image-preview"
     GEMINI_2_5_PRO = "gemini-2.5-pro"
     GEMINI_2_5_FLASH = "gemini-2.5-flash"
+    GEMINI_2_5_FLASH_IMAGE_PREVIEW = "gemini-2.5-flash-image-preview"
+    GEMINI_3_PRO_PREVIEW = "gemini-3-pro-preview"
+    GEMINI_3_PRO_IMAGE_PREVIEW = "gemini-3-pro-image-preview"
+    GEMINI_3_FLASH_PREVIEW = "gemini-3-flash-preview"
     VTO = "virtual-try-on-preview-08-04"
 
     # Video-Specific Models

backend/src/common/base_repository.py

@@ -24,11 +24,15 @@
 # Define generic types for SQLAlchemy Model and Pydantic Schema
 ModelType = TypeVar("ModelType", bound=Any)
 SchemaType = TypeVar("SchemaType", bound=BaseModel)
+IDType = TypeVar("IDType", bound=Union[int, str])
 
 
-class BaseDocument(BaseModel):
-    # ID is an int because it might be auto-generated by the DB (Integer).
-    id: int = None
+class BaseDocumentMixin(BaseModel, Generic[IDType]):
+    """
+    Base Pydantic model for all schemas.
+    Now generic over IDType to enforce strict int or str IDs.
+    """
+    id: IDType
     created_at: datetime.datetime = Field(
         default_factory=lambda: datetime.datetime.now(datetime.timezone.utc)
     )
@@ -46,17 +50,28 @@ class BaseDocument(BaseModel):
     )
 
 
-class BaseRepository(Generic[ModelType, SchemaType]):
+# Backward compatibility: BaseDocument defaults to int IDs
+class BaseDocument(BaseDocumentMixin[int]):
+    pass
+
+
+# New base class for models that require String IDs (e.g. UUIDs)
+class BaseStringDocument(BaseDocumentMixin[str]):
+    pass
+
+
+class BaseRepositoryMixin(Generic[ModelType, SchemaType, IDType]):
     """
-    A generic repository for common SQLAlchemy operations.
+    A generic repository mixin for common SQLAlchemy operations.
+    Generic over IDType to enforce strict types.
     """
 
     def __init__(self, model: Type[ModelType], schema: Type[SchemaType], db: AsyncSession):
         self.model = model
         self.schema = schema
         self.db = db
 
-    async def get_by_id(self, item_id: int) -> Optional[SchemaType]:
+    async def get_by_id(self, item_id: IDType) -> Optional[SchemaType]:
         """Retrieves a single document by its ID."""
         result = await self.db.execute(
             select(self.model).where(self.model.id == item_id)
@@ -79,14 +94,14 @@ async def create(self, schema: Union[BaseModel, Dict[str, Any]]) -> SchemaType:
         # We exclude 'id' if it's None so the DB can auto-increment it (for Int IDs)
         if data.get("id") is None:
             data.pop("id", None)
-            
+
         db_item = self.model(**data)
         self.db.add(db_item)
         await self.db.commit()
         await self.db.refresh(db_item)
         return self.schema.model_validate(db_item)
 
-    async def update(self, item_id: int, update_data: Union[BaseModel, Dict[str, Any]]) -> Optional[SchemaType]:
+    async def update(self, item_id: IDType, update_data: Union[BaseModel, Dict[str, Any]]) -> Optional[SchemaType]:
         """
         Performs a partial update on a document.
         """
@@ -108,7 +123,7 @@ async def update(self, item_id: int, update_data: Union[BaseModel, Dict[str, Any
         for key, value in data.items():
             if hasattr(db_item, key):
                 setattr(db_item, key, value)
-        
+
         # Update timestamp
         if hasattr(db_item, "updated_at"):
             db_item.updated_at = datetime.datetime.now(datetime.timezone.utc)
@@ -117,7 +132,7 @@ async def update(self, item_id: int, update_data: Union[BaseModel, Dict[str, Any
         await self.db.refresh(db_item)
         return self.schema.model_validate(db_item)
 
-    async def delete(self, item_id: int) -> bool:
+    async def delete(self, item_id: IDType) -> bool:
         """
         Deletes a document by its ID.
         Returns True if deletion was successful (item existed), False otherwise.
@@ -127,7 +142,7 @@ async def delete(self, item_id: int) -> bool:
             delete(self.model).where(self.model.id == item_id)
         )
         await self.db.commit()
-        return result.rowcount > 0
+        return result.rowcount > 0 # type: ignore
 
     async def find_all(self, limit: int = 100, offset: int = 0) -> List[SchemaType]:
         """
@@ -138,3 +153,13 @@ async def find_all(self, limit: int = 100, offset: int = 0) -> List[SchemaType]:
         )
         items = result.scalars().all()
         return [self.schema.model_validate(item) for item in items]
+
+
+# BaseRepository defaults to int IDs
+class BaseRepository(BaseRepositoryMixin[ModelType, SchemaType, int]):
+    pass
+
+
+# BaseStringRepository requires String IDs (e.g. UUIDs)
+class BaseStringRepository(BaseRepositoryMixin[ModelType, SchemaType, str]):
+    pass

backend/src/config/config_service.py

@@ -41,6 +41,7 @@ class ConfigService(BaseSettings):
     LOCATION: str = "global"
     ENVIRONMENT: str = "development"
     FRONTEND_URL: str = "http://localhost:4200"
+    BACKEND_URL: str = "http://localhost:8080"
     LOG_LEVEL: str = "INFO"
     INIT_VERTEX: bool = True
 
@@ -94,6 +95,13 @@ class ConfigService(BaseSettings):
     )
     ADMIN_USER_EMAIL: str = "system"
 
+    # --- Workflows ---
+    WORKFLOWS_LOCATION: str = "us-central1"
+    WORKFLOWS_EXECUTOR_URL: str = (
+        "http://localhost:8080"  # This service could be deployed alone in the future
+    )
+    BACKEND_SERVICE_ACCOUNT_EMAIL: str = ""
+
     @model_validator(mode="before")
     @classmethod
     def get_default_project_id(cls, values: Any) -> Any:

backend/src/galleries/gallery_service.py

@@ -70,7 +70,7 @@ async def _enrich_source_asset_link(
         Fetches the source asset document and generates a presigned URL for it.
         """
         asset_doc = await self.source_asset_repo.get_by_id(link.asset_id)
-        
+
         if not asset_doc:
             return None
 
@@ -259,7 +259,7 @@ async def get_paginated_gallery(
         )
 
     async def get_media_by_id(
-        self, item_id: str, current_user: UserModel
+        self, item_id: int, current_user: UserModel
     ) -> Optional[MediaItemResponse]:
         """
         Retrieves a single media item, performs an authorization check,