|
5 | 5 | package io.airbyte.server.handlers; |
6 | 6 |
|
7 | 7 | import com.google.common.annotations.VisibleForTesting; |
| 8 | +import com.google.common.base.Preconditions; |
8 | 9 | import com.google.common.collect.ImmutableMap; |
9 | 10 | import com.google.common.collect.ImmutableMap.Builder; |
10 | 11 | import com.google.common.collect.Lists; |
@@ -116,7 +117,7 @@ public ConnectionRead createConnection(final ConnectionCreate connectionCreate) |
116 | 117 | ConnectionHelper.validateWorkspace(workspaceHelper, |
117 | 118 | connectionCreate.getSourceId(), |
118 | 119 | connectionCreate.getDestinationId(), |
119 | | - new HashSet<>(operationIds)); |
| 120 | + operationIds); |
120 | 121 |
|
121 | 122 | final UUID connectionId = uuidGenerator.get(); |
122 | 123 |
|
@@ -229,26 +230,123 @@ private Builder<String, Object> generateMetadata(final StandardSync standardSync |
229 | 230 | return metadata; |
230 | 231 | } |
231 | 232 |
|
232 | | - public ConnectionRead updateConnection(final ConnectionUpdate connectionUpdate) |
| 233 | + public ConnectionRead updateConnection(final ConnectionUpdate connectionPatch) |
233 | 234 | throws ConfigNotFoundException, IOException, JsonValidationException { |
234 | | - // retrieve and update sync |
235 | | - final StandardSync persistedSync = configRepository.getStandardSync(connectionUpdate.getConnectionId()); |
236 | 235 |
|
237 | | - final StandardSync newConnection = ConnectionHelper.updateConnectionObject( |
238 | | - workspaceHelper, |
239 | | - persistedSync, |
240 | | - ApiPojoConverters.connectionUpdateToInternal(connectionUpdate)); |
241 | | - ConnectionHelper.validateWorkspace( |
242 | | - workspaceHelper, |
243 | | - persistedSync.getSourceId(), |
244 | | - persistedSync.getDestinationId(), |
245 | | - new HashSet<>(connectionUpdate.getOperationIds())); |
| 236 | + final UUID connectionId = connectionPatch.getConnectionId(); |
| 237 | + |
| 238 | + LOGGER.debug("Starting updateConnection for connectionId {}...", connectionId); |
| 239 | + LOGGER.debug("incoming connectionPatch: {}", connectionPatch); |
| 240 | + |
| 241 | + final StandardSync sync = configRepository.getStandardSync(connectionId); |
| 242 | + LOGGER.debug("initial StandardSync: {}", sync); |
| 243 | + |
| 244 | + validateConnectionPatch(workspaceHelper, sync, connectionPatch); |
| 245 | + |
| 246 | + final ConnectionRead initialConnectionRead = ApiPojoConverters.internalToConnectionRead(sync); |
| 247 | + LOGGER.debug("initial ConnectionRead: {}", initialConnectionRead); |
| 248 | + |
| 249 | + applyPatchToStandardSync(sync, connectionPatch); |
| 250 | + |
| 251 | + LOGGER.debug("patched StandardSync before persisting: {}", sync); |
| 252 | + configRepository.writeStandardSync(sync); |
| 253 | + |
| 254 | + eventRunner.update(connectionId); |
| 255 | + |
| 256 | + final ConnectionRead updatedRead = buildConnectionRead(connectionId); |
| 257 | + LOGGER.debug("final connectionRead: {}", updatedRead); |
| 258 | + |
| 259 | + return updatedRead; |
| 260 | + } |
| 261 | + |
| 262 | + /** |
| 263 | + * Modifies the given StandardSync by applying changes from a partially-filled ConnectionUpdate |
| 264 | + * patch. Any fields that are null in the patch will be left unchanged. |
| 265 | + */ |
| 266 | + private static void applyPatchToStandardSync(final StandardSync sync, final ConnectionUpdate patch) throws JsonValidationException { |
| 267 | + // update the sync's schedule using the patch's scheduleType and scheduleData. validations occur in |
| 268 | + // the helper to ensure both fields |
| 269 | + // make sense together. |
| 270 | + if (patch.getScheduleType() != null) { |
| 271 | + ConnectionScheduleHelper.populateSyncFromScheduleTypeAndData(sync, patch.getScheduleType(), patch.getScheduleData()); |
| 272 | + } |
| 273 | + |
| 274 | + // the rest of the fields are straightforward to patch. If present in the patch, set the field to |
| 275 | + // the value |
| 276 | + // in the patch. Otherwise, leave the field unchanged. |
| 277 | + |
| 278 | + if (patch.getSyncCatalog() != null) { |
| 279 | + sync.setCatalog(CatalogConverter.toProtocol(patch.getSyncCatalog())); |
| 280 | + } |
246 | 281 |
|
247 | | - configRepository.writeStandardSync(newConnection); |
| 282 | + if (patch.getName() != null) { |
| 283 | + sync.setName(patch.getName()); |
| 284 | + } |
| 285 | + |
| 286 | + if (patch.getNamespaceDefinition() != null) { |
| 287 | + sync.setNamespaceDefinition(Enums.convertTo(patch.getNamespaceDefinition(), NamespaceDefinitionType.class)); |
| 288 | + } |
| 289 | + |
| 290 | + if (patch.getNamespaceFormat() != null) { |
| 291 | + sync.setNamespaceFormat(patch.getNamespaceFormat()); |
| 292 | + } |
| 293 | + |
| 294 | + if (patch.getPrefix() != null) { |
| 295 | + sync.setPrefix(patch.getPrefix()); |
| 296 | + } |
| 297 | + |
| 298 | + if (patch.getOperationIds() != null) { |
| 299 | + sync.setOperationIds(patch.getOperationIds()); |
| 300 | + } |
| 301 | + |
| 302 | + if (patch.getStatus() != null) { |
| 303 | + sync.setStatus(ApiPojoConverters.toPersistenceStatus(patch.getStatus())); |
| 304 | + } |
| 305 | + |
| 306 | + if (patch.getSourceCatalogId() != null) { |
| 307 | + sync.setSourceCatalogId(patch.getSourceCatalogId()); |
| 308 | + } |
| 309 | + |
| 310 | + if (patch.getResourceRequirements() != null) { |
| 311 | + sync.setResourceRequirements(ApiPojoConverters.resourceRequirementsToInternal(patch.getResourceRequirements())); |
| 312 | + } |
| 313 | + } |
248 | 314 |
|
249 | | - eventRunner.update(connectionUpdate.getConnectionId()); |
| 315 | + private void validateConnectionPatch(final WorkspaceHelper workspaceHelper, final StandardSync persistedSync, final ConnectionUpdate patch) { |
| 316 | + // sanity check that we're updating the right connection |
| 317 | + Preconditions.checkArgument(persistedSync.getConnectionId().equals(patch.getConnectionId())); |
250 | 318 |
|
251 | | - return buildConnectionRead(connectionUpdate.getConnectionId()); |
| 319 | + // make sure all operationIds belong to the same workspace as the connection |
| 320 | + ConnectionHelper.validateWorkspace( |
| 321 | + workspaceHelper, persistedSync.getSourceId(), persistedSync.getDestinationId(), patch.getOperationIds()); |
| 322 | + |
| 323 | + // make sure the incoming schedule update is sensible. Note that schedule details are further |
| 324 | + // validated in ConnectionScheduleHelper, this just |
| 325 | + // sanity checks that fields are populated when they should be. |
| 326 | + Preconditions.checkArgument( |
| 327 | + patch.getSchedule() == null, |
| 328 | + "ConnectionUpdate should only make changes to the schedule by setting scheduleType and scheduleData. 'schedule' is no longer supported."); |
| 329 | + |
| 330 | + if (patch.getScheduleType() == null) { |
| 331 | + Preconditions.checkArgument( |
| 332 | + patch.getScheduleData() == null, |
| 333 | + "ConnectionUpdate should not include any scheduleData without also specifying a valid scheduleType."); |
| 334 | + } else { |
| 335 | + switch (patch.getScheduleType()) { |
| 336 | + case MANUAL -> Preconditions.checkArgument( |
| 337 | + patch.getScheduleData() == null, |
| 338 | + "ConnectionUpdate should not include any scheduleData when setting the Connection scheduleType to MANUAL."); |
| 339 | + case BASIC -> Preconditions.checkArgument( |
| 340 | + patch.getScheduleData() != null, |
| 341 | + "ConnectionUpdate should include scheduleData when setting the Connection scheduleType to BASIC."); |
| 342 | + case CRON -> Preconditions.checkArgument( |
| 343 | + patch.getScheduleData() != null, |
| 344 | + "ConnectionUpdate should include scheduleData when setting the Connection scheduleType to CRON."); |
| 345 | + |
| 346 | + // shouldn't be possible to reach this case |
| 347 | + default -> throw new RuntimeException("Unrecognized scheduleType!"); |
| 348 | + } |
| 349 | + } |
252 | 350 | } |
253 | 351 |
|
254 | 352 | public ConnectionReadList listConnectionsForWorkspace(final WorkspaceIdRequestBody workspaceIdRequestBody) |
|
0 commit comments