Application Factory

def create_app_from_workspace_process_context(
    workspace_process_context: IWorkspaceProcessContext,
    path_prefix: str = "",
    live_data_poll_rate: Optional[int] = None,
    **kwargs
) -> Starlette:
    """
    Create a Starlette ASGI application from workspace process context.
    
    Args:
        workspace_process_context: The workspace context containing instance and code locations
        path_prefix: URL path prefix for hosting (must start with '/' if provided)
        live_data_poll_rate: Rate in milliseconds for UI to poll for live data updates
        **kwargs: Additional arguments passed to Starlette constructor
    
    Returns:
        Starlette: ASGI application ready for deployment
        
    Raises:
        Exception: If path_prefix doesn't start with '/' or ends with '/'
    """

from dagster import DagsterInstance
from dagster._core.workspace.context import WorkspaceProcessContext
from dagster_webserver.app import create_app_from_workspace_process_context
import uvicorn

# Basic usage
instance = DagsterInstance.get()
with WorkspaceProcessContext(instance) as workspace_context:
    app = create_app_from_workspace_process_context(workspace_context)
    uvicorn.run(app, host="127.0.0.1", port=3000)

# With path prefix for reverse proxy
with WorkspaceProcessContext(instance) as workspace_context:
    app = create_app_from_workspace_process_context(
        workspace_context,
        path_prefix="/dagster"
    )
    uvicorn.run(app, host="0.0.0.0", port=8080)

# With custom live data polling
with WorkspaceProcessContext(instance) as workspace_context:
    app = create_app_from_workspace_process_context(
        workspace_context,
        live_data_poll_rate=5000  # 5 second polling
    )
    uvicorn.run(app)

# With custom Starlette options
with WorkspaceProcessContext(instance) as workspace_context:
    app = create_app_from_workspace_process_context(
        workspace_context,
        debug=True,
        middleware=[custom_middleware]
    )

# Gunicorn deployment
# gunicorn -w 4 -k uvicorn.workers.UvicornWorker myapp:app

# FastAPI integration
from fastapi import FastAPI
from fastapi.middleware.wsgi import WSGIMiddleware

main_app = FastAPI()
dagster_app = create_app_from_workspace_process_context(workspace_context, path_prefix="/dagster")
main_app.mount("/dagster", dagster_app)

# Custom ASGI server
import hypercorn.asyncio
import hypercorn.config

config = hypercorn.config.Config()
config.bind = ["0.0.0.0:3000"]
asyncio.run(hypercorn.asyncio.serve(app, config))

from dagster._core.workspace.context import WorkspaceProcessContext
from dagster_webserver.debug import WebserverDebugWorkspaceProcessContext

# Standard workspace context
instance = DagsterInstance.get()
with WorkspaceProcessContext(
    instance,
    workspace_load_target=workspace_load_target,
    version="1.11.8"
) as workspace_context:
    app = create_app_from_workspace_process_context(workspace_context)

# Debug workspace context with ephemeral instance
debug_instance = DagsterInstance.ephemeral(preload=debug_payloads)
with WebserverDebugWorkspaceProcessContext(debug_instance) as debug_context:
    app = create_app_from_workspace_process_context(debug_context)

# Read-only workspace context
with WorkspaceProcessContext(
    instance,
    workspace_load_target=workspace_load_target,
    read_only=True
) as readonly_context:
    app = create_app_from_workspace_process_context(readonly_context)

# Valid path prefix examples
app = create_app_from_workspace_process_context(workspace_context, path_prefix="/dagster")
app = create_app_from_workspace_process_context(workspace_context, path_prefix="/my-org/dagster-ui")

# Invalid path prefixes (will raise exceptions)
# path_prefix="dagster"        # Missing leading slash
# path_prefix="/dagster/"      # Trailing slash not allowed

# Fast polling for development (1 second)
app = create_app_from_workspace_process_context(
    workspace_context,
    live_data_poll_rate=1000
)

# Slow polling for resource conservation (10 seconds) 
app = create_app_from_workspace_process_context(
    workspace_context,
    live_data_poll_rate=10000
)

# Default polling (2 seconds)
app = create_app_from_workspace_process_context(workspace_context)

# Dockerfile application
from dagster import DagsterInstance
from dagster._core.workspace.context import WorkspaceProcessContext  
from dagster_webserver.app import create_app_from_workspace_process_context

instance = DagsterInstance.from_config("/app/dagster.yaml")
with WorkspaceProcessContext(instance) as workspace_context:
    app = create_app_from_workspace_process_context(
        workspace_context,
        path_prefix=os.getenv("DAGSTER_PATH_PREFIX", "")
    )

if __name__ == "__main__":
    import uvicorn
    uvicorn.run("app:app", host="0.0.0.0", port=3000)

# Health check endpoint can be added via custom middleware
from starlette.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware

class HealthCheckMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        if request.url.path == "/health":
            return JSONResponse({"status": "healthy"})
        return await call_next(request)

app = create_app_from_workspace_process_context(
    workspace_context,
    middleware=[HealthCheckMiddleware]
)

# Multiple instances with different configurations
apps = {}

for config_name, instance_config in instance_configs.items():
    instance = DagsterInstance.from_config(instance_config)
    with WorkspaceProcessContext(instance) as workspace_context:
        apps[config_name] = create_app_from_workspace_process_context(
            workspace_context,
            path_prefix=f"/{config_name}"
        )

# Mount in main application
from starlette.applications import Starlette
from starlette.routing import Mount

main_app = Starlette(routes=[
    Mount(f"/{name}", app) for name, app in apps.items()
])

# Path prefix validation
try:
    app = create_app_from_workspace_process_context(
        workspace_context,
        path_prefix="invalid-prefix"  # Missing leading slash
    )
except Exception as e:
    print(f"Configuration error: {e}")

# Workspace context validation  
try:
    app = create_app_from_workspace_process_context(None)
except Exception as e:
    print(f"Invalid workspace context: {e}")

# Custom middleware integration
custom_middleware = [
    Middleware(CustomAuthMiddleware),
    Middleware(CORSMiddleware, allow_origins=["*"])
]

app = create_app_from_workspace_process_context(
    workspace_context,
    middleware=custom_middleware
)

# Exception handling
from starlette.exceptions import HTTPException
from starlette.responses import JSONResponse

async def http_exception_handler(request, exc):
    return JSONResponse(
        {"error": str(exc.detail)},
        status_code=exc.status_code
    )

app = create_app_from_workspace_process_context(
    workspace_context,
    exception_handlers={HTTPException: http_exception_handler}
)