DynamodDb access in Java(Code)

Assume we have below data in DynamoDb. All fields are String type. PK is ID, SK is name.

We use DynamoDbEnhancedAsyncClient to access DynamoDb data.

To get a specific record with (PK, SK):

public static void testGetItem() throws Exception {
    CompletableFuture<Record> recordFuture = table.getItem(Key.builder()
            .partitionValue("001")
            .sortValue(33)
            .build());
    System.out.println(recordFuture.get());
}

To query only on PK:

public static void testQueryNormal() throws Exception {
    QueryConditional queryConditional = QueryConditional.keyEqualTo(Key.builder()
            .partitionValue("003")
            .build());
    PagePublisher<Record> pagePublisher = table.query(queryConditional);
    Flux<Page<Record>> recordFlux = Flux.from(pagePublisher);
    recordFlux.subscribe(record -> {
        System.out.println(record.items());
    });
    Thread.sleep(2000l);
}

To query on (PK, SK prefix):

public static void testQuerySomeSortKey() throws Exception {
    QueryConditional queryConditional = QueryConditional.sortBeginsWith(Key.builder()
            .partitionValue("002")
            .sortValue("pp")
            .build());
    PagePublisher<Record> pagePublisher = table.query(queryConditional);
    Flux<Page<Record>> recordFlux = Flux.from(pagePublisher);
    recordFlux.subscribe(record -> {
        System.out.println(record.items());
        System.out.println(record.lastEvaluatedKey());
    });
    Thread.sleep(2000l);
}

To query with pagination:

public Mono<RecordWithLastKey> queryPagination1stCall() throws Exception {
    QueryConditional queryConditional = QueryConditional.keyEqualTo(Key.builder()
            .partitionValue("003")
            .build());
    QueryEnhancedRequest enhancedRequest = QueryEnhancedRequest.builder()
            .queryConditional(queryConditional)
            .limit(2)
            .build();
    PagePublisher<Record> pagePublisher = table.query(enhancedRequest);
    Flux<Page<Record>> recordFlux = Flux.from(pagePublisher);

    return recordFlux.next()
            .map(record -> new RecordWithLastKey(record.items(), getRecordKey(record.lastEvaluatedKey())));
}

public Mono<RecordWithLastKey> queryPaginationNextCall(String pk, String sk) throws Exception {
    QueryConditional queryConditional = QueryConditional.keyEqualTo(Key.builder()
            .partitionValue("003")
            .build());
    Map<String, AttributeValue> lastEvaluatedKey = new HashMap<>();
    lastEvaluatedKey.put("ID", AttributeValue.builder().s(pk).build());
    lastEvaluatedKey.put("NAME", AttributeValue.builder().s(sk).build());
    QueryEnhancedRequest enhancedRequest = QueryEnhancedRequest.builder()
            .queryConditional(queryConditional)
            .exclusiveStartKey(lastEvaluatedKey)
            .limit(2)
            .build();

    PagePublisher<Record> pagePublisher = table.query(enhancedRequest);
    Flux<Page<Record>> recordFlux = Flux.from(pagePublisher);

    return recordFlux.next()
            .map(record -> new RecordWithLastKey(record.items(), getRecordKey(record.lastEvaluatedKey())));
}

private RecordKey getRecordKey(Map<String, AttributeValue> valueMap) {
    RecordKey recordKey = null;
    if (valueMap != null) {
        recordKey = RecordKey.builder()
                .pk(valueMap.get("ID").s())
                .sk(valueMap.get("NAME").s())
                .build();
    }
    return recordKey;
}

see code on github.