0
# Application Factory
1

2
Core functionality for creating ASGI applications from workspace contexts. Enables programmatic webserver integration, custom deployment scenarios, and embedding the Dagster UI in larger applications.
3

4
## Capabilities
5

6
### Application Creation
7

8
Main factory function for creating Starlette ASGI applications with full Dagster webserver functionality.
9

10
```python { .api }
11
def create_app_from_workspace_process_context(
12
    workspace_process_context: IWorkspaceProcessContext,
13
    path_prefix: str = "",
14
    live_data_poll_rate: Optional[int] = None,
15
    **kwargs
16
) -> Starlette:
17
    """
18
    Create a Starlette ASGI application from workspace process context.
19
    
20
    Args:
21
        workspace_process_context: The workspace context containing instance and code locations
22
        path_prefix: URL path prefix for hosting (must start with '/' if provided)
23
        live_data_poll_rate: Rate in milliseconds for UI to poll for live data updates
24
        **kwargs: Additional arguments passed to Starlette constructor
25
    
26
    Returns:
27
        Starlette: ASGI application ready for deployment
28
        
29
    Raises:
30
        Exception: If path_prefix doesn't start with '/' or ends with '/'
31
    """
32
```
33

34
**Usage Examples:**
35

36
```python
37
from dagster import DagsterInstance
38
from dagster._core.workspace.context import WorkspaceProcessContext
39
from dagster_webserver.app import create_app_from_workspace_process_context
40
import uvicorn
41

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

48
# With path prefix for reverse proxy
49
with WorkspaceProcessContext(instance) as workspace_context:
50
    app = create_app_from_workspace_process_context(
51
        workspace_context,
52
        path_prefix="/dagster"
53
    )
54
    uvicorn.run(app, host="0.0.0.0", port=8080)
55

56
# With custom live data polling
57
with WorkspaceProcessContext(instance) as workspace_context:
58
    app = create_app_from_workspace_process_context(
59
        workspace_context,
60
        live_data_poll_rate=5000  # 5 second polling
61
    )
62
    uvicorn.run(app)
63

64
# With custom Starlette options
65
with WorkspaceProcessContext(instance) as workspace_context:
66
    app = create_app_from_workspace_process_context(
67
        workspace_context,
68
        debug=True,
69
        middleware=[custom_middleware]
70
    )
71
```
72

73
### ASGI Integration
74

75
The created application is a standard ASGI app compatible with various deployment scenarios:
76

77
```python
78
# Gunicorn deployment
79
# gunicorn -w 4 -k uvicorn.workers.UvicornWorker myapp:app
80

81
# FastAPI integration
82
from fastapi import FastAPI
83
from fastapi.middleware.wsgi import WSGIMiddleware
84

85
main_app = FastAPI()
86
dagster_app = create_app_from_workspace_process_context(workspace_context, path_prefix="/dagster")
87
main_app.mount("/dagster", dagster_app)
88

89
# Custom ASGI server
90
import hypercorn.asyncio
91
import hypercorn.config
92

93
config = hypercorn.config.Config()
94
config.bind = ["0.0.0.0:3000"]
95
asyncio.run(hypercorn.asyncio.serve(app, config))
96
```
97

98
### Workspace Context Management
99

100
The application factory works with various workspace context types:
101

102
```python
103
from dagster._core.workspace.context import WorkspaceProcessContext
104
from dagster_webserver.debug import WebserverDebugWorkspaceProcessContext
105

106
# Standard workspace context
107
instance = DagsterInstance.get()
108
with WorkspaceProcessContext(
109
    instance,
110
    workspace_load_target=workspace_load_target,
111
    version="1.11.8"
112
) as workspace_context:
113
    app = create_app_from_workspace_process_context(workspace_context)
114

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

120
# Read-only workspace context
121
with WorkspaceProcessContext(
122
    instance,
123
    workspace_load_target=workspace_load_target,
124
    read_only=True
125
) as readonly_context:
126
    app = create_app_from_workspace_process_context(readonly_context)
127
```
128

129
### Path Prefix Configuration
130

131
Path prefixes enable hosting Dagster UI under subpaths for reverse proxy scenarios:
132

133
```python
134
# Valid path prefix examples
135
app = create_app_from_workspace_process_context(workspace_context, path_prefix="/dagster")
136
app = create_app_from_workspace_process_context(workspace_context, path_prefix="/my-org/dagster-ui")
137

138
# Invalid path prefixes (will raise exceptions)
139
# path_prefix="dagster"        # Missing leading slash
140
# path_prefix="/dagster/"      # Trailing slash not allowed
141
```
142

143
### Live Data Configuration
144

145
Control how frequently the UI polls for live data updates:
146

147
```python
148
# Fast polling for development (1 second)
149
app = create_app_from_workspace_process_context(
150
    workspace_context,
151
    live_data_poll_rate=1000
152
)
153

154
# Slow polling for resource conservation (10 seconds) 
155
app = create_app_from_workspace_process_context(
156
    workspace_context,
157
    live_data_poll_rate=10000
158
)
159

160
# Default polling (2 seconds)
161
app = create_app_from_workspace_process_context(workspace_context)
162
```
163

164
## Deployment Patterns
165

166
### Container Deployment
167

168
```python
169
# Dockerfile application
170
from dagster import DagsterInstance
171
from dagster._core.workspace.context import WorkspaceProcessContext  
172
from dagster_webserver.app import create_app_from_workspace_process_context
173

174
instance = DagsterInstance.from_config("/app/dagster.yaml")
175
with WorkspaceProcessContext(instance) as workspace_context:
176
    app = create_app_from_workspace_process_context(
177
        workspace_context,
178
        path_prefix=os.getenv("DAGSTER_PATH_PREFIX", "")
179
    )
180

181
if __name__ == "__main__":
182
    import uvicorn
183
    uvicorn.run("app:app", host="0.0.0.0", port=3000)
184
```
185

186
### Kubernetes Deployment
187

188
```python
189
# Health check endpoint can be added via custom middleware
190
from starlette.responses import JSONResponse
191
from starlette.middleware.base import BaseHTTPMiddleware
192

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

199
app = create_app_from_workspace_process_context(
200
    workspace_context,
201
    middleware=[HealthCheckMiddleware]
202
)
203
```
204

205
### Multi-Instance Deployment
206

207
```python
208
# Multiple instances with different configurations
209
apps = {}
210

211
for config_name, instance_config in instance_configs.items():
212
    instance = DagsterInstance.from_config(instance_config)
213
    with WorkspaceProcessContext(instance) as workspace_context:
214
        apps[config_name] = create_app_from_workspace_process_context(
215
            workspace_context,
216
            path_prefix=f"/{config_name}"
217
        )
218

219
# Mount in main application
220
from starlette.applications import Starlette
221
from starlette.routing import Mount
222

223
main_app = Starlette(routes=[
224
    Mount(f"/{name}", app) for name, app in apps.items()
225
])
226
```
227

228
## Error Handling
229

230
The application factory performs validation and provides clear error messages:
231

232
```python
233
# Path prefix validation
234
try:
235
    app = create_app_from_workspace_process_context(
236
        workspace_context,
237
        path_prefix="invalid-prefix"  # Missing leading slash
238
    )
239
except Exception as e:
240
    print(f"Configuration error: {e}")
241

242
# Workspace context validation  
243
try:
244
    app = create_app_from_workspace_process_context(None)
245
except Exception as e:
246
    print(f"Invalid workspace context: {e}")
247
```
248

249
## Integration Helpers
250

251
The application supports various integration scenarios through additional configuration:
252

253
```python
254
# Custom middleware integration
255
custom_middleware = [
256
    Middleware(CustomAuthMiddleware),
257
    Middleware(CORSMiddleware, allow_origins=["*"])
258
]
259

260
app = create_app_from_workspace_process_context(
261
    workspace_context,
262
    middleware=custom_middleware
263
)
264

265
# Exception handling
266
from starlette.exceptions import HTTPException
267
from starlette.responses import JSONResponse
268

269
async def http_exception_handler(request, exc):
270
    return JSONResponse(
271
        {"error": str(exc.detail)},
272
        status_code=exc.status_code
273
    )
274

275
app = create_app_from_workspace_process_context(
276
    workspace_context,
277
    exception_handlers={HTTPException: http_exception_handler}
278
)
279
```