Skip to content

Commit 701e065

Browse files
committed
wip of adding database as part of job records; tests
1 parent 9bd2d70 commit 701e065

11 files changed

Lines changed: 573 additions & 175 deletions

File tree

AGENTS.md

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,13 @@
1+
# App Engine Pipelines
2+
3+
This project is an advanced, modernized implementation of Google App Engine Pipelines. Originally an execution framework to run complex, multi-step asynchronous algorithms on App Engine, this fork extends the concepts and capabilities to be compatible with newer Java versions, modern GCP Datastore API clients (Cloud Datastore / Firestore in Datastore mode), and Cloud Tasks.
4+
5+
## Project Architecture
6+
- **Datastore Client**: Uses modern `google-cloud-datastore` APIs rather than the legacy `appengine-api-1.0-sdk`.
7+
- **Task Queues**: Abstracts task enqueuing to work across `AppEngineTaskQueue` (legacy standard App Engine push queues) and `CloudTasksTaskQueue` (modern Google Cloud Tasks).
8+
- **Settings Propagation**: Inherits pipeline settings such as retry counts, worker services, worker versions, and custom Datastore boundaries (Namespace and Database ID) down to sub-jobs and async worker callbacks.
9+
10+
## Development Rules
11+
- When modifying datastore interactions, always ensure you respect potential overrides for `databaseId` and `namespace`, which are typically configured at the `JobSetting` level.
12+
- When passing parameters to new Pipeline Tasks, leverage `PipelineTask.toProperties()` and augment `QueueSettings` if those parameters must be inherited across jobs.
13+
- Validate Cloud Datastore constraints carefully (e.g., namespace regex restrictions).

java/src/main/java/com/google/appengine/tools/pipeline/JobSetting.java

Lines changed: 38 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -14,15 +14,13 @@
1414

1515
package com.google.appengine.tools.pipeline;
1616

17-
import lombok.Getter;
18-
import lombok.NonNull;
19-
import lombok.RequiredArgsConstructor;
20-
2117
import java.io.Serial;
2218
import java.io.Serializable;
2319
import java.util.Arrays;
2420
import java.util.Optional;
2521

22+
import lombok.Getter;
23+
import lombok.RequiredArgsConstructor;
2624

2725
/**
2826
* A setting for specifying to the framework some aspect of a Job's execution.
@@ -77,7 +75,7 @@ abstract class StringValuedSetting implements JobSetting {
7775
@Serial
7876
private static final long serialVersionUID = 7756646651569386669L;
7977

80-
//NOTE: behavior of Pipeline Framework allows this to be null for some settings
78+
// NOTE: behavior of Pipeline Framework allows this to be null for some settings
8179
// (tests verify this)
8280
private final String value;
8381

@@ -179,8 +177,9 @@ public OnService(String service) {
179177
*/
180178
final class OnServiceVersion extends StringValuedSetting {
181179

182-
@Serial
180+
@Serial
183181
private static final long serialVersionUID = 3877411731586475273L;
182+
184183
public OnServiceVersion(String version) {
185184
super(version);
186185
}
@@ -212,20 +211,49 @@ public StatusConsoleUrl(String statusConsoleUrl) {
212211
}
213212
}
214213

214+
/**
215+
* A setting for specifying the datastore database to use for this job;
216+
* otherwise will be the default datastore database.
217+
*
218+
* q: do we want to allow pipelines to mix datastore databases?
219+
*
220+
*/
221+
final class DatastoreDatabase extends StringValuedSetting {
222+
@Serial
223+
private static final long serialVersionUID = -1L;
224+
225+
public DatastoreDatabase(String datastoreDatabase) {
226+
super(datastoreDatabase);
227+
if (datastoreDatabase != null && !datastoreDatabase.isEmpty() && !datastoreDatabase.equals("(default)")) {
228+
if (!datastoreDatabase.matches("^[a-z][a-z0-9-]{1,61}[a-z0-9]$")) {
229+
throw new IllegalArgumentException("Invalid Datastore database ID: " + datastoreDatabase);
230+
}
231+
}
232+
}
233+
}
234+
235+
/**
236+
* A setting for specifying the datastore namespace to use for this job;
237+
* otherwise will be the default datastore namespace.
238+
*/
215239
final class DatastoreNamespace extends StringValuedSetting {
216240
@Serial
217241
private static final long serialVersionUID = -1L;
218242

219243
public DatastoreNamespace(String datastoreNameSpace) {
220244
super(datastoreNameSpace);
245+
if (datastoreNameSpace != null) {
246+
if (!datastoreNameSpace.matches("^[0-9A-Za-z._-]{0,100}$")) {
247+
throw new IllegalArgumentException("Invalid Datastore namespace: " + datastoreNameSpace);
248+
}
249+
}
221250
}
222251
}
223252

224-
225253
static <E extends StringValuedSetting> Optional<String> getSettingValue(Class<E> clazz, JobSetting[] settings) {
226254
return Arrays.stream(settings)
227-
.filter( s -> s.getClass().isAssignableFrom(clazz))
228-
.findAny()
229-
.map(s -> ((StringValuedSetting) s).getValue());
255+
.filter(s -> s.getClass().isAssignableFrom(clazz))
256+
.findAny()
257+
.map(s -> ((StringValuedSetting) s).getValue());
230258
}
231259
}

java/src/main/java/com/google/appengine/tools/pipeline/impl/QueueSettings.java

Lines changed: 31 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,14 @@
11
package com.google.appengine.tools.pipeline.impl;
22

3-
import lombok.*;
4-
53
import javax.annotation.Nullable;
64

5+
import lombok.AllArgsConstructor;
6+
import lombok.Builder;
7+
import lombok.Getter;
8+
import lombok.NoArgsConstructor;
9+
import lombok.Setter;
10+
import lombok.ToString;
11+
712
/**
813
* settings for how to asynchronously execute a task via a queue
914
*
@@ -12,7 +17,8 @@
1217
@AllArgsConstructor
1318
@NoArgsConstructor
1419
@Builder
15-
@Getter @Setter
20+
@Getter
21+
@Setter
1622
@ToString
1723
public final class QueueSettings implements Cloneable {
1824

@@ -33,12 +39,26 @@ public final class QueueSettings implements Cloneable {
3339
private String onQueue;
3440

3541
/**
36-
* delay in seconds to set when enqueueing the task (eg, should not execute until *at least* this much time has passed
42+
* delay in seconds to set when enqueueing the task (eg, should not execute
43+
* until *at least* this much time has passed
3744
*/
3845
private Long delayInSeconds;
3946

4047
/**
41-
* Merge will override any {@code null} setting with a matching setting from {@code other}.
48+
* datastore database ID to propagate
49+
*/
50+
@Nullable
51+
private String databaseId;
52+
53+
/**
54+
* datastore namespace to propagate
55+
*/
56+
@Nullable
57+
private String namespace;
58+
59+
/**
60+
* Merge will override any {@code null} setting with a matching setting from
61+
* {@code other}.
4262
* Note, delay value is not being merged.
4363
*/
4464
public QueueSettings merge(QueueSettings other) {
@@ -49,6 +69,12 @@ public QueueSettings merge(QueueSettings other) {
4969
if (onQueue == null) {
5070
onQueue = other.getOnQueue();
5171
}
72+
if (databaseId == null) {
73+
databaseId = other.getDatabaseId();
74+
}
75+
if (namespace == null) {
76+
namespace = other.getNamespace();
77+
}
5278
return this;
5379
}
5480

0 commit comments

Comments
 (0)