Add spring-batch-notion

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/Filter.java

@@ -0,0 +1,615 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import notion.api.v1.model.databases.query.filter.CompoundFilterElement;
+import notion.api.v1.model.databases.query.filter.QueryTopLevelFilter;
+import notion.api.v1.model.databases.query.filter.condition.CheckboxFilter;
+import notion.api.v1.model.databases.query.filter.condition.MultiSelectFilter;
+import notion.api.v1.model.databases.query.filter.condition.NumberFilter;
+import notion.api.v1.model.databases.query.filter.condition.SelectFilter;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.function.BiConsumer;
+import java.util.function.BiFunction;
+import java.util.function.Consumer;
+
+/**
+ * Filtering conditions to limit the entries returned from a database query.
+ * <p>
+ * Filters operate on property values or entry timestamps, and can be combined.
+ * <p>
+ * A filter definition starts with {@link #where()}, entry point for a fluent API that
+ * mimics the database filter option in the Notion UI.
+ *
+ * @author Stefano Cordio
+ */
+public abstract sealed class Filter {
+
+	/**
+	 * Entry point that starts the definition of a filter.
+	 * @return a new {@link FilterConditionBuilder} instance
+	 */
+	public static FilterConditionBuilder<TopLevelFilter> where() {
+		return new FilterConditionBuilder<>(PropertyFilter::new);
+	}
+
+	/**
+	 * Entry point that starts the definition of a filter group.
+	 * @param filter the filter representing the group
+	 * @return a new {@link TopLevelFilter} instance that delegates to the given filter
+	 */
+	public static TopLevelFilter where(Filter filter) {
+		return new DelegateFilter(filter);
+	}
+
+	private Filter() {
+	}
+
+	abstract QueryTopLevelFilter toQueryTopLevelFilter();
+
+	abstract CompoundFilterElement toCompoundFilterElement();
+
+	/**
+	 * Base class for top level filters that support filters composition via the
+	 * {@link TopLevelFilter#and} and {@link TopLevelFilter#or} methods.
+	 *
+	 * @see AndFilter
+	 * @see OrFilter
+	 */
+	public static abstract sealed class TopLevelFilter extends Filter {
+
+		private TopLevelFilter() {
+		}
+
+		/**
+		 * Start the definition of a new filter that is composed with the current filter
+		 * via a logical {@code and}.
+		 * @return a {@link FilterConditionBuilder} instance for an {@link AndFilter}
+		 */
+		public FilterConditionBuilder<AndFilter> and() {
+			return new FilterConditionBuilder<>(
+					(property, customizer) -> new AndFilter(this, new PropertyFilter(property, customizer)));
+		}
+
+		/**
+		 * Compose the current filter and the given filter via a logical {@code and}.
+		 * @param filter the filter to compose with the current filter
+		 * @return a new {@link AndFilter} instance
+		 */
+		public AndFilter and(Filter filter) {
+			return new AndFilter(this, Objects.requireNonNull(filter));
+		}
+
+		/**
+		 * Start the definition of a new filter that is composed with the current filter
+		 * via a logical {@code or}.
+		 * @return a {@link FilterConditionBuilder} instance for an {@link OrFilter}
+		 */
+		public FilterConditionBuilder<OrFilter> or() {
+			return new FilterConditionBuilder<>(
+					(property, customizer) -> new OrFilter(this, new PropertyFilter(property, customizer)));
+		}
+
+		/**
+		 * Compose the current filter and the given filter via a logical {@code or}.
+		 * @param filter the filter to compose with the current filter
+		 * @return a new {@link OrFilter} instance
+		 */
+		public OrFilter or(Filter filter) {
+			return new OrFilter(this, Objects.requireNonNull(filter));
+		}
+
+	}
+
+	private static final class DelegateFilter extends TopLevelFilter {
+
+		private final Filter delegate;
+
+		private DelegateFilter(Filter delegate) {
+			this.delegate = Objects.requireNonNull(delegate);
+		}
+
+		@Override
+		QueryTopLevelFilter toQueryTopLevelFilter() {
+			return delegate.toQueryTopLevelFilter();
+		}
+
+		@Override
+		CompoundFilterElement toCompoundFilterElement() {
+			return delegate.toCompoundFilterElement();
+		}
+
+	}
+
+	private static final class PropertyFilter extends TopLevelFilter {
+
+		private final String property;
+
+		private final NotionPropertyFilterCustomizer customizer;
+
+		private PropertyFilter(String property, NotionPropertyFilterCustomizer customizer) {
+			this.property = property;
+			this.customizer = customizer;
+		}
+
+		@Override
+		QueryTopLevelFilter toQueryTopLevelFilter() {
+			return toNotionPropertyFilter();
+		}
+
+		@Override
+		CompoundFilterElement toCompoundFilterElement() {
+			return toNotionPropertyFilter();
+		}
+
+		private notion.api.v1.model.databases.query.filter.PropertyFilter toNotionPropertyFilter() {
+			var notionPropertyFilter = new notion.api.v1.model.databases.query.filter.PropertyFilter(property);
+			customizer.accept(notionPropertyFilter);
+			return notionPropertyFilter;
+		}
+
+	}
+
+	@FunctionalInterface
+	private interface NotionPropertyFilterFactory<T extends Filter>
+			extends BiFunction<String, NotionPropertyFilterCustomizer, T> {
+
+	}
+
+	@FunctionalInterface
+	private interface NotionPropertyFilterCustomizer
+			extends Consumer<notion.api.v1.model.databases.query.filter.PropertyFilter> {
+
+	}
+
+	static abstract sealed class CompoundFilter extends Filter {
+
+		final List<Filter> filters = new ArrayList<>();
+
+		private final NotionCompoundFilterSetter setter;
+
+		private CompoundFilter(NotionCompoundFilterSetter setter) {
+			this.setter = setter;
+		}
+
+		@Override
+		QueryTopLevelFilter toQueryTopLevelFilter() {
+			return toNotionCompoundFilter();
+		}
+
+		@Override
+		CompoundFilterElement toCompoundFilterElement() {
+			return toNotionCompoundFilter();
+		}
+
+		private notion.api.v1.model.databases.query.filter.CompoundFilter toNotionCompoundFilter() {
+			var notionCompoundFilter = new notion.api.v1.model.databases.query.filter.CompoundFilter();
+			var notionCompoundFilterElements = filters.stream().map(Filter::toCompoundFilterElement).toList();
+			setter.accept(notionCompoundFilter, notionCompoundFilterElements);
+			return notionCompoundFilter;
+		}
+
+	}
+
+	@FunctionalInterface
+	private interface NotionCompoundFilterSetter
+			extends BiConsumer<notion.api.v1.model.databases.query.filter.CompoundFilter, List<CompoundFilterElement>> {
+
+	}
+
+	/**
+	 * Compound filter that supports filters composition via the {@link AndFilter#and}
+	 * methods.
+	 * <p>
+	 * Returns entries that match <b>all</b> of the provided filters.
+	 */
+	public static final class AndFilter extends CompoundFilter {
+
+		private AndFilter(Filter first, Filter second) {
+			super(notion.api.v1.model.databases.query.filter.CompoundFilter::setAnd);
+			filters.addAll(List.of(first, second));
+		}
+
+		/**
+		 * Start the definition of a new filter that is composed with the current filter
+		 * via a logical {@code and}.
+		 * @return a {@link FilterConditionBuilder} instance for an {@link AndFilter}
+		 */
+		public FilterConditionBuilder<AndFilter> and() {
+			return new FilterConditionBuilder<>((property, customizer) -> {
+				filters.add(new PropertyFilter(property, customizer));
+				return this;
+			});
+		}
+
+		/**
+		 * Compose the current filter and the given filter via a logical {@code and}.
+		 * @param filter the filter to compose with the current filter
+		 * @return a new {@link AndFilter} instance
+		 */
+		public AndFilter and(Filter filter) {
+			filters.add(Objects.requireNonNull(filter));
+			return this;
+		}
+
+	}
+
+	/**
+	 * Compound filter that supports filters composition via the {@link OrFilter#or}
+	 * methods.
+	 * <p>
+	 * Returns entries that match <b>any</b> of the provided filters.
+	 */
+	public static final class OrFilter extends CompoundFilter {
+
+		private OrFilter(Filter first, Filter second) {
+			super(notion.api.v1.model.databases.query.filter.CompoundFilter::setOr);
+			filters.addAll(List.of(first, second));
+		}
+
+		/**
+		 * Start the definition of a new filter that is composed with the current filter
+		 * via a logical {@code or}.
+		 * @return a {@link FilterConditionBuilder} instance for an {@link OrFilter}
+		 */
+		public FilterConditionBuilder<OrFilter> or() {
+			return new FilterConditionBuilder<>((property, customizer) -> {
+				filters.add(new PropertyFilter(property, customizer));
+				return this;
+			});
+		}
+
+		/**
+		 * Compose the current filter and the given filter via a logical {@code or}.
+		 * @param filter the filter to compose with the current filter
+		 * @return a new {@link OrFilter} instance
+		 */
+		public OrFilter or(Filter filter) {
+			filters.add(Objects.requireNonNull(filter));
+			return this;
+		}
+
+	}
+
+	/**
+	 * Builder for {@link Filter} conditions.
+	 *
+	 * @param <T> the type of the target {@code Filter}
+	 */
+	public static final class FilterConditionBuilder<T extends Filter> {
+
+		private final NotionPropertyFilterFactory<T> factory;
+
+		private FilterConditionBuilder(NotionPropertyFilterFactory<T> factory) {
+			this.factory = factory;
+		}
+
+		/**
+		 * Start the definition of the filter condition for a {@code checkbox} property.
+		 * @param property The name of the property as it appears in the database, or the
+		 * property ID
+		 * @return a new {@link CheckboxCondition} instance
+		 */
+		public CheckboxCondition<T> checkbox(String property) {
+			return new CheckboxCondition<>(property, factory);
+		}
+
+		/**
+		 * Start the definition of the filter condition for a {@code multi-select}
+		 * property.
+		 * @param property The name of the property as it appears in the database, or the
+		 * property ID
+		 * @return a new {@link MultiSelectCondition} instance
+		 */
+		public MultiSelectCondition<T> multiSelect(String property) {
+			return new MultiSelectCondition<>(property, factory);
+		}
+
+		/**
+		 * Start the definition of the filter condition for a {@code number} property.
+		 * @param property The name of the property as it appears in the database, or the
+		 * property ID
+		 * @return a new {@link NumberCondition} instance
+		 */
+		public NumberCondition<T> number(String property) {
+			return new NumberCondition<>(property, factory);
+		}
+
+		/**
+		 * Start the definition of the filter condition for a {@code select} property.
+		 * @param property The name of the property as it appears in the database, or the
+		 * property ID
+		 * @return a new {@link SelectCondition} instance
+		 */
+		public SelectCondition<T> select(String property) {
+			return new SelectCondition<>(property, factory);
+		}
+
+		static abstract sealed class Condition<T extends Filter> {
+
+			private final String property;
+
+			private final NotionPropertyFilterFactory<T> factory;
+
+			private Condition(String property, NotionPropertyFilterFactory<T> factory) {
+				this.property = property;
+				this.factory = factory;
+			}
+
+			T toFilter(NotionPropertyFilterCustomizer customizer) {
+				return factory.apply(property, customizer);
+			}
+
+		}
+
+		/**
+		 * Filter condition for a {@code checkbox} property.
+		 *
+		 * @param <T> the type of the target filter
+		 */
+		public static final class CheckboxCondition<T extends Filter> extends Condition<T> {
+
+			private CheckboxCondition(String property, NotionPropertyFilterFactory<T> factory) {
+				super(property, factory);
+			}
+
+			/**
+			 * Return all database entries with an exact value match.
+			 * @param value the value to match
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEqualTo(boolean value) {
+				CheckboxFilter checkboxFilter = new CheckboxFilter();
+				checkboxFilter.setEquals(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setCheckbox(checkboxFilter));
+			}
+
+			/**
+			 * Return all database entries without an exact value match.
+			 * @param value the value to differ with
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEqualTo(boolean value) {
+				CheckboxFilter checkboxFilter = new CheckboxFilter();
+				checkboxFilter.setDoesNotEqual(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setCheckbox(checkboxFilter));
+			}
+
+		}
+
+		/**
+		 * Filter condition for a {@code multi-select} property.
+		 *
+		 * @param <T> the type of the target filter
+		 */
+		public static final class MultiSelectCondition<T extends Filter> extends Condition<T> {
+
+			private MultiSelectCondition(String property, NotionPropertyFilterFactory<T> factory) {
+				super(property, factory);
+			}
+
+			/**
+			 * Return database entries where the provided value is part of the property
+			 * values.
+			 * @param value the value to compare the property values against
+			 * @return a filter with the newly defined condition
+			 */
+			public T contains(String value) {
+				MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+				multiSelectFilter.setContains(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setMultiSelect(multiSelectFilter));
+			}
+
+			/**
+			 * Return database entries where the provided value is not contained in the
+			 * property values.
+			 * @param value the value to compare the property values against
+			 * @return a filter with the newly defined condition
+			 */
+			public T doesNotContain(String value) {
+				MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+				multiSelectFilter.setDoesNotContain(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setMultiSelect(multiSelectFilter));
+			}
+
+			/**
+			 * Return database entries where the property value does not contain any data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEmpty() {
+				MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+				multiSelectFilter.setEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setMultiSelect(multiSelectFilter));
+			}
+
+			/**
+			 * Return database entries where the property value contains data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEmpty() {
+				MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+				multiSelectFilter.setNotEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setMultiSelect(multiSelectFilter));
+			}
+
+		}
+
+		/**
+		 * Filter condition for a {@code number} property.
+		 *
+		 * @param <T> the type of the target filter
+		 */
+		public static final class NumberCondition<T extends Filter> extends Condition<T> {
+
+			private NumberCondition(String property, NotionPropertyFilterFactory<T> factory) {
+				super(property, factory);
+			}
+
+			/**
+			 * Return database entries where the property value is the same as the
+			 * provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEqualTo(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setEquals(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value differs from the provided
+			 * one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEqualTo(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setDoesNotEqual(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value exceeds the provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isGreaterThan(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setGreaterThan(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value is equal to or exceeds the
+			 * provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isGreaterThanOrEqualTo(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setGreaterThanOrEqualTo(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value is less than the provided
+			 * one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isLessThan(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setLessThan(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value is equal to or is less
+			 * than the provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isLessThanOrEqualTo(int value) {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setLessThanOrEqualTo(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value does not contain any data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEmpty() {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+			/**
+			 * Return database entries where the property value contains data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEmpty() {
+				NumberFilter numberFilter = new NumberFilter();
+				numberFilter.setNotEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setNumber(numberFilter));
+			}
+
+		}
+
+		/**
+		 * Filter condition for a {@code select} property.
+		 *
+		 * @param <T> the type of the target filter
+		 */
+		public static final class SelectCondition<T extends Filter> extends Condition<T> {
+
+			private SelectCondition(String property, NotionPropertyFilterFactory<T> factory) {
+				super(property, factory);
+			}
+
+			/**
+			 * Return database entries where the property value matches the provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEqualTo(String value) {
+				SelectFilter selectFilter = new SelectFilter();
+				selectFilter.setEquals(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setSelect(selectFilter));
+			}
+
+			/**
+			 * Return database entries where the property value does not match the
+			 * provided one.
+			 * @param value the value to compare the property value against
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEqualTo(String value) {
+				SelectFilter selectFilter = new SelectFilter();
+				selectFilter.setDoesNotEqual(value);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setSelect(selectFilter));
+			}
+
+			/**
+			 * Return database entries where the property value does not contain any data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isEmpty() {
+				SelectFilter selectFilter = new SelectFilter();
+				selectFilter.setEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setSelect(selectFilter));
+			}
+
+			/**
+			 * Return database entries where the property value contains data.
+			 * @return a filter with the newly defined condition
+			 */
+			public T isNotEmpty() {
+				SelectFilter selectFilter = new SelectFilter();
+				selectFilter.setNotEmpty(true);
+				return toFilter(notionPropertyFilter -> notionPropertyFilter.setSelect(selectFilter));
+			}
+
+		}
+
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/NotionDatabaseItemReader.java

@@ -0,0 +1,267 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import org.springframework.batch.extensions.notion.mapping.PropertyMapper;
+import notion.api.v1.NotionClient;
+import notion.api.v1.http.JavaNetHttpClient;
+import notion.api.v1.logging.Slf4jLogger;
+import notion.api.v1.model.databases.QueryResults;
+import notion.api.v1.model.databases.query.filter.QueryTopLevelFilter;
+import notion.api.v1.model.databases.query.sort.QuerySort;
+import notion.api.v1.model.pages.Page;
+import notion.api.v1.model.pages.PageProperty;
+import notion.api.v1.model.pages.PageProperty.RichText;
+import notion.api.v1.request.databases.QueryDatabaseRequest;
+import org.springframework.batch.item.ExecutionContext;
+import org.springframework.batch.item.ItemReader;
+import org.springframework.batch.item.data.AbstractPaginatedDataItemReader;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.util.Assert;
+
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Restartable {@link ItemReader} that reads entries from a Notion database via a paging
+ * technique.
+ * <p>
+ * The query is executed using paged requests of a size specified in
+ * {@link #setPageSize(int)}, which defaults to {@value #DEFAULT_PAGE_SIZE}. Additional
+ * pages are requested as needed when the {@link #read()} method is called. On restart,
+ * the reader will begin again at the same number item it left off at.
+ * <p>
+ * This implementation is thread-safe between calls to {@link #open(ExecutionContext)},
+ * but remember to set <code>saveState</code> to <code>false</code> if used in a
+ * multi-threaded environment (no restart available).
+ *
+ * @author Stefano Cordio
+ * @param <T> Type of item to be read
+ */
+public class NotionDatabaseItemReader<T> extends AbstractPaginatedDataItemReader<T> implements InitializingBean {
+
+	private static final String DEFAULT_BASE_URL = "https://api.notion.com/v1";
+
+	private static final int DEFAULT_PAGE_SIZE = 100;
+
+	private String baseUrl;
+
+	private String token;
+
+	private String databaseId;
+
+	private PropertyMapper<T> propertyMapper;
+
+	private QueryTopLevelFilter filter;
+
+	private List<QuerySort> sorts;
+
+	private NotionClient client;
+
+	private boolean hasMore;
+
+	private String nextCursor;
+
+	/**
+	 * Create a new {@link NotionDatabaseItemReader} with the following defaults:
+	 * <ul>
+	 * <li>{@code baseUrl} = {@value #DEFAULT_BASE_URL}</li>
+	 * <li>{@code pageSize} = {@value #DEFAULT_PAGE_SIZE}</li>
+	 * </ul>
+	 */
+	public NotionDatabaseItemReader() {
+		this.baseUrl = DEFAULT_BASE_URL;
+		this.pageSize = DEFAULT_PAGE_SIZE;
+	}
+
+	/**
+	 * The base URL of the Notion API.
+	 * <p>
+	 * Defaults to {@value #DEFAULT_BASE_URL}.
+	 * <p>
+	 * A custom value can be provided for testing purposes (e.g., the URL of a WireMock
+	 * server).
+	 * @param baseUrl the base URL
+	 */
+	public void setBaseUrl(String baseUrl) {
+		this.baseUrl = Objects.requireNonNull(baseUrl);
+	}
+
+	/**
+	 * The Notion integration token.
+	 * <p>
+	 * Always required.
+	 * @param token the token
+	 */
+	public void setToken(String token) {
+		this.token = Objects.requireNonNull(token);
+	}
+
+	/**
+	 * UUID of the database to read from.
+	 * <p>
+	 * Always required.
+	 * @param databaseId the database UUID
+	 */
+	public void setDatabaseId(String databaseId) {
+		this.databaseId = Objects.requireNonNull(databaseId);
+	}
+
+	/**
+	 * The {@link PropertyMapper} responsible for mapping Notion item properties into a
+	 * Java object.
+	 * <p>
+	 * Always required.
+	 * @param propertyMapper the property mapper
+	 */
+	public void setPropertyMapper(PropertyMapper<T> propertyMapper) {
+		this.propertyMapper = Objects.requireNonNull(propertyMapper);
+	}
+
+	/**
+	 * {@link Filter} condition to limit the returned items.
+	 * <p>
+	 * If no filter is provided, all the items in the database will be returned.
+	 * @param filter the {@link Filter} conditions
+	 * @see Filter#where()
+	 * @see Filter#where(Filter)
+	 */
+	public void setFilter(Filter filter) {
+		this.filter = filter.toQueryTopLevelFilter();
+	}
+
+	/**
+	 * {@link Sort} conditions to order the returned items.
+	 * <p>
+	 * Each condition is applied following the declaration order, i.e., earlier sorts take
+	 * precedence over later ones.
+	 * @param sorts the {@link Sort} conditions
+	 * @see Sort#by(String)
+	 * @see Sort#by(Sort.Timestamp)
+	 */
+	public void setSorts(Sort... sorts) {
+		this.sorts = Stream.of(sorts).map(Sort::toQuerySort).toList();
+	}
+
+	/**
+	 * The number of items to be read with each page.
+	 * <p>
+	 * Defaults to {@value #DEFAULT_PAGE_SIZE}.
+	 * @param pageSize the number of items. Must be greater than 0 and less than or equal
+	 * to 100.
+	 */
+	@Override
+	public void setPageSize(int pageSize) {
+		Assert.isTrue(pageSize <= 100, "pageSize must be less than or equal to 100");
+		super.setPageSize(pageSize);
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	protected Iterator<T> doPageRead() {
+		if (!hasMore) {
+			return null;
+		}
+
+		QueryDatabaseRequest request = new QueryDatabaseRequest(databaseId);
+		request.setFilter(filter);
+		request.setSorts(sorts);
+		request.setStartCursor(nextCursor);
+		request.setPageSize(pageSize);
+
+		QueryResults queryResults = client.queryDatabase(request);
+
+		hasMore = queryResults.getHasMore();
+		nextCursor = queryResults.getNextCursor();
+
+		return queryResults.getResults()
+			.stream()
+			.map(NotionDatabaseItemReader::getProperties)
+			.map(properties -> propertyMapper.map(properties))
+			.iterator();
+	}
+
+	private static Map<String, String> getProperties(Page element) {
+		return element.getProperties()
+			.entrySet()
+			.stream()
+			.collect(Collectors.toUnmodifiableMap(Entry::getKey, entry -> getPropertyValue(entry.getValue())));
+	}
+
+	private static String getPropertyValue(PageProperty property) {
+		return switch (property.getType()) {
+			case RichText -> getPlainText(property.getRichText());
+			case Title -> getPlainText(property.getTitle());
+			default -> throw new IllegalArgumentException("Unsupported type: " + property.getType());
+		};
+	}
+
+	private static String getPlainText(List<RichText> texts) {
+		return texts.isEmpty() ? "" : texts.get(0).getPlainText();
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	protected void doOpen() {
+		client = new NotionClient(token);
+		client.setHttpClient(new JavaNetHttpClient());
+		client.setLogger(new Slf4jLogger());
+		client.setBaseUrl(baseUrl);
+
+		hasMore = true;
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	protected void doClose() {
+		client.close();
+		client = null;
+
+		hasMore = false;
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	protected void jumpToItem(int itemIndex) throws Exception {
+		for (int i = 0; i < itemIndex; i++) {
+			read();
+		}
+	}
+
+	/**
+	 * {@inheritDoc}
+	 */
+	@Override
+	public void afterPropertiesSet() {
+		Assert.state(token != null, "'token' must be set");
+		Assert.state(databaseId != null, "'databaseId' must be set");
+		Assert.state(propertyMapper != null, "'propertyMapper' must be set");
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/Sort.java

@@ -0,0 +1,186 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import notion.api.v1.model.databases.query.sort.QuerySort;
+import notion.api.v1.model.databases.query.sort.QuerySortDirection;
+import notion.api.v1.model.databases.query.sort.QuerySortTimestamp;
+
+import java.util.Objects;
+
+/**
+ * Sort conditions to order the entries returned from a database query.
+ * <p>
+ * Sorts operate on property values or entry timestamps, and can be combined.
+ * <p>
+ * The direction defaults to {@link Direction#DEFAULT_DIRECTION}.
+ *
+ * @author Stefano Cordio
+ */
+public abstract sealed class Sort {
+
+	/**
+	 * Default direction of {@link Sort} conditions.
+	 */
+	public static final Direction DEFAULT_DIRECTION = Direction.ASCENDING;
+
+	/**
+	 * Sort condition that orders the database query by a particular property.
+	 * @param property the name of the property to sort against
+	 * @param direction the {@code Direction} to sort
+	 * @return the {@code Sort} instance
+	 */
+	public static Sort by(String property, Direction direction) {
+		return new PropertySort(property, direction);
+	}
+
+	/**
+	 * Sort condition that orders the database query by a particular property, in
+	 * ascending direction.
+	 * @param property the name of the property to sort against
+	 * @return the {@code Sort} instance
+	 */
+	public static Sort by(String property) {
+		return new PropertySort(property, DEFAULT_DIRECTION);
+	}
+
+	/**
+	 * Sort condition that orders the database query by the timestamp associated with a
+	 * database entry.
+	 * @param timestamp the {@code Timestamp} to sort against
+	 * @param direction the {@code Direction} to sort
+	 * @return the {@code Sort} instance
+	 */
+	public static Sort by(Timestamp timestamp, Direction direction) {
+		return new TimestampSort(timestamp, direction);
+	}
+
+	/**
+	 * Sort condition that orders the database query by the timestamp associated with a
+	 * database entry, in ascending direction.
+	 * @param timestamp the {@code Timestamp timestamp} to sort against
+	 * @return the {@code Sort} instance
+	 */
+	public static Sort by(Timestamp timestamp) {
+		return new TimestampSort(timestamp, DEFAULT_DIRECTION);
+	}
+
+	/**
+	 * Timestamps associated with database entries.
+	 */
+	public enum Timestamp {
+
+		/**
+		 * The time the entry was created.
+		 */
+		CREATED_TIME(QuerySortTimestamp.CreatedTime),
+
+		/**
+		 * The time the entry was last edited.
+		 */
+		LAST_EDITED_TIME(QuerySortTimestamp.LastEditedTime);
+
+		private final QuerySortTimestamp querySortTimestamp;
+
+		Timestamp(QuerySortTimestamp querySortTimestamp) {
+			this.querySortTimestamp = querySortTimestamp;
+		}
+
+		private QuerySortTimestamp getQuerySortTimestamp() {
+			return querySortTimestamp;
+		}
+
+	}
+
+	/**
+	 * Sort directions.
+	 */
+	public enum Direction {
+
+		/**
+		 * Ascending direction.
+		 */
+		ASCENDING(QuerySortDirection.Ascending),
+
+		/**
+		 * Descending direction.
+		 */
+		DESCENDING(QuerySortDirection.Descending);
+
+		private final QuerySortDirection querySortDirection;
+
+		Direction(QuerySortDirection querySortDirection) {
+			this.querySortDirection = querySortDirection;
+		}
+
+		private QuerySortDirection getQuerySortDirection() {
+			return querySortDirection;
+		}
+
+	}
+
+	private Sort() {
+	}
+
+	abstract QuerySort toQuerySort();
+
+	private static final class PropertySort extends Sort {
+
+		private final String property;
+
+		private final Direction direction;
+
+		private PropertySort(String property, Direction direction) {
+			this.property = Objects.requireNonNull(property);
+			this.direction = Objects.requireNonNull(direction);
+		}
+
+		@Override
+		QuerySort toQuerySort() {
+			return new QuerySort(property, null, direction.getQuerySortDirection());
+		}
+
+		@Override
+		public String toString() {
+			return "%s: %s".formatted(property, direction);
+		}
+
+	}
+
+	private static final class TimestampSort extends Sort {
+
+		private final Timestamp timestamp;
+
+		private final Direction direction;
+
+		private TimestampSort(Timestamp timestamp, Direction direction) {
+			this.timestamp = Objects.requireNonNull(timestamp);
+			this.direction = Objects.requireNonNull(direction);
+		}
+
+		@Override
+		QuerySort toQuerySort() {
+			return new QuerySort(null, timestamp.getQuerySortTimestamp(), direction.getQuerySortDirection());
+		}
+
+		@Override
+		public String toString() {
+			return "%s: %s".formatted(timestamp, direction);
+		}
+
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/BeanWrapperPropertyMapper.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.beans.BeanWrapper;
+import org.springframework.beans.PropertyAccessorFactory;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+
+import java.lang.reflect.Constructor;
+
+/**
+ * {@link PropertyMapper} implementation for JavaBeans.
+ * <p>
+ * It requires a default constructor and expects the setter names to match the Notion item
+ * property names (case-insensitive).
+ *
+ * @author Stefano Cordio
+ * @param <T> the target type
+ */
+public class BeanWrapperPropertyMapper<T> extends CaseInsensitivePropertyMapper<T> {
+
+	private final Constructor<T> constructor;
+
+	/**
+	 * Create a new {@link BeanWrapperPropertyMapper} for the given target type.
+	 * @param type type of the target object
+	 */
+	public BeanWrapperPropertyMapper(Class<T> type) {
+		this.constructor = BeanUtils.getResolvableConstructor(type);
+	}
+
+	/**
+	 * Create a new {@link BeanWrapperPropertyMapper}, inferring the target type.
+	 * @param reified don't pass any values to it. It's a trick to detect the target type.
+	 */
+	@SafeVarargs
+	public BeanWrapperPropertyMapper(T... reified) {
+		this(ClassResolver.getClassOf(reified));
+	}
+
+	@Override
+	T mapCaseInsensitive(LinkedCaseInsensitiveMap<String> properties) {
+		T instance = BeanUtils.instantiateClass(constructor);
+		BeanWrapper beanWrapper = PropertyAccessorFactory.forBeanPropertyAccess(instance);
+		beanWrapper.setPropertyValues(properties);
+		return instance;
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/CaseInsensitivePropertyMapper.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.util.LinkedCaseInsensitiveMap;
+
+import java.util.Map;
+
+/**
+ * @author Stefano Cordio
+ */
+abstract class CaseInsensitivePropertyMapper<T> implements PropertyMapper<T> {
+
+	@Override
+	public T map(Map<String, String> properties) {
+		LinkedCaseInsensitiveMap<String> caseInsensitiveProperties = new LinkedCaseInsensitiveMap<>(properties.size());
+		caseInsensitiveProperties.putAll(properties);
+		return mapCaseInsensitive(caseInsensitiveProperties);
+	}
+
+	abstract T mapCaseInsensitive(LinkedCaseInsensitiveMap<String> properties);
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/ClassResolver.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.util.Assert;
+
+/**
+ * @author Stefano Cordio
+ */
+class ClassResolver {
+
+	@SuppressWarnings("unchecked")
+	static <T> Class<T> getClassOf(T[] reified) {
+		Assert.isTrue(reified.length == 0,
+				"Please don't pass any values here. The type will be detected automagically.");
+
+		return (Class<T>) reified.getClass().getComponentType();
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/ConstructorBasedPropertyMapper.java

@@ -0,0 +1,54 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.beans.BeanUtils;
+import org.springframework.util.LinkedCaseInsensitiveMap;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.Parameter;
+import java.util.Arrays;
+
+/**
+ * @author Stefano Cordio
+ */
+abstract class ConstructorBasedPropertyMapper<T> extends CaseInsensitivePropertyMapper<T> {
+
+	private final Constructor<T> constructor;
+
+	ConstructorBasedPropertyMapper(Class<T> type) {
+		try {
+			this.constructor = getConstructor(type);
+		}
+		catch (NoSuchMethodException e) {
+			throw new IllegalArgumentException(e);
+		}
+	}
+
+	abstract Constructor<T> getConstructor(Class<T> type) throws NoSuchMethodException;
+
+	@Override
+	T mapCaseInsensitive(LinkedCaseInsensitiveMap<String> properties) {
+		Object[] parameterValues = Arrays.stream(constructor.getParameters()) //
+			.map(Parameter::getName) //
+			.map(properties::get) //
+			.toArray();
+
+		return BeanUtils.instantiateClass(constructor, parameterValues);
+
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/ConstructorPropertyMapper.java

@@ -0,0 +1,65 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import java.lang.reflect.Constructor;
+import java.util.Arrays;
+
+/**
+ * {@link PropertyMapper} implementation for types with a constructor with arguments.
+ * <p>
+ * It requires the constructor to be unique and its parameter names to match the Notion
+ * item property names (case-insensitive).
+ *
+ * @author Stefano Cordio
+ * @param <T> the target type
+ */
+public class ConstructorPropertyMapper<T> extends ConstructorBasedPropertyMapper<T> {
+
+	/**
+	 * Create a new {@link ConstructorPropertyMapper} for the given target type.
+	 * @param type type of the target object
+	 */
+	public ConstructorPropertyMapper(Class<T> type) {
+		super(type);
+	}
+
+	/**
+	 * Create a new {@link ConstructorPropertyMapper}, inferring the target type.
+	 * @param reified don't pass any values to it. It's a trick to detect the target type.
+	 */
+	@SafeVarargs
+	public ConstructorPropertyMapper(T... reified) {
+		this(ClassResolver.getClassOf(reified));
+	}
+
+	@SuppressWarnings("unchecked")
+	@Override
+	Constructor<T> getConstructor(Class<T> type) throws NoSuchMethodException {
+		Constructor<?>[] constructors = type.getDeclaredConstructors();
+
+		if (constructors.length == 0) {
+			throw new NoSuchMethodException("No constructor found for type: " + type);
+		}
+
+		if (constructors.length > 1) {
+			throw new NoSuchMethodException("Multiple constructors available: " + Arrays.toString(constructors));
+		}
+
+		return (Constructor<T>) constructors[0];
+	}
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/PropertyMapper.java

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import java.util.Map;
+
+/**
+ * Strategy interface for mapping the properties of a Notion item into a Java object.
+ *
+ * @author Stefano Cordio
+ * @param <T> the object type
+ */
+@FunctionalInterface
+public interface PropertyMapper<T> {
+
+	/**
+	 * Map the given item properties into an object of type {@code T}.
+	 * @param properties unmodifiable map containing the property value objects, keyed by
+	 * property name
+	 * @return the populated object
+	 */
+	T map(Map<String, String> properties);
+
+}

spring-batch-notion/src/main/java/org/springframework/batch/extensions/notion/mapping/RecordPropertyMapper.java

@@ -0,0 +1,61 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.util.ReflectionUtils;
+
+import java.lang.reflect.Constructor;
+import java.lang.reflect.RecordComponent;
+import java.util.Arrays;
+
+/**
+ * {@link PropertyMapper} implementation for {@link Record Java records}.
+ * <p>
+ * It uses the record's canonical constructor and requires the record's component names to
+ * match the Notion item property names (case-insensitive).
+ *
+ * @author Stefano Cordio
+ * @param <T> the target type &mdash; must be a {@link Record}
+ */
+public class RecordPropertyMapper<T extends Record> extends ConstructorBasedPropertyMapper<T> {
+
+	/**
+	 * Create a new {@link RecordPropertyMapper} for the given target type.
+	 * @param type type of the target record
+	 */
+	public RecordPropertyMapper(Class<T> type) {
+		super(type);
+	}
+
+	/**
+	 * Create a new {@link RecordPropertyMapper}, inferring the target type.
+	 * @param reified don't pass any values to it. It's a trick to detect the target type.
+	 */
+	@SafeVarargs
+	public RecordPropertyMapper(T... reified) {
+		this(ClassResolver.getClassOf(reified));
+	}
+
+	@Override
+	Constructor<T> getConstructor(Class<T> type) throws NoSuchMethodException {
+		Class<?>[] parameterTypes = Arrays.stream(type.getRecordComponents()) //
+			.map(RecordComponent::getType) //
+			.toArray(Class[]::new);
+
+		return ReflectionUtils.accessibleConstructor(type, parameterTypes);
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/FilterTests.java

@@ -0,0 +1,529 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import notion.api.v1.model.databases.query.filter.CompoundFilter;
+import notion.api.v1.model.databases.query.filter.PropertyFilter;
+import notion.api.v1.model.databases.query.filter.QueryTopLevelFilter;
+import notion.api.v1.model.databases.query.filter.condition.CheckboxFilter;
+import notion.api.v1.model.databases.query.filter.condition.MultiSelectFilter;
+import notion.api.v1.model.databases.query.filter.condition.NumberFilter;
+import notion.api.v1.model.databases.query.filter.condition.SelectFilter;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.List;
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import static org.springframework.batch.extensions.notion.Filter.where;
+import static java.util.function.Function.identity;
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+/**
+ * @author Stefano Cordio
+ */
+class FilterTests {
+
+	@ParameterizedTest
+	@MethodSource({ "propertyFilters", "compoundFilters", "nestedFilters" })
+	void toQueryTopLevelFilter(Filter underTest, QueryTopLevelFilter expected) {
+		// WHEN
+		QueryTopLevelFilter result = underTest.toQueryTopLevelFilter();
+		// THEN
+		then(result).usingRecursiveComparison().isEqualTo(expected);
+	}
+
+	static Stream<Arguments> propertyFilters() {
+		return Stream.of( //
+				checkboxFilters(), //
+				multiSelectFilters(), //
+				numberFilters(), //
+				selectFilters()) //
+			.flatMap(identity());
+	}
+
+	static Stream<Arguments> checkboxFilters() {
+		return Stream.of(true, false)
+			.flatMap(value -> Stream.of( //
+					arguments( //
+							where().checkbox("property").isEqualTo(value), //
+							supply(() -> {
+								CheckboxFilter checkboxFilter = new CheckboxFilter();
+								checkboxFilter.setEquals(value);
+								PropertyFilter propertyFilter = new PropertyFilter("property");
+								propertyFilter.setCheckbox(checkboxFilter);
+								return propertyFilter;
+							})),
+					arguments( //
+							where().checkbox("property").isNotEqualTo(value), //
+							supply(() -> {
+								CheckboxFilter checkboxFilter = new CheckboxFilter();
+								checkboxFilter.setDoesNotEqual(value);
+								PropertyFilter propertyFilter = new PropertyFilter("property");
+								propertyFilter.setCheckbox(checkboxFilter);
+								return propertyFilter;
+							}))));
+	}
+
+	static Stream<Arguments> multiSelectFilters() {
+		return Stream.of( //
+				arguments( //
+						where().multiSelect("property").contains("value"), //
+						supply(() -> {
+							MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+							multiSelectFilter.setContains("value");
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setMultiSelect(multiSelectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().multiSelect("property").doesNotContain("value"), //
+						supply(() -> {
+							MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+							multiSelectFilter.setDoesNotContain("value");
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setMultiSelect(multiSelectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().multiSelect("property").isEmpty(), //
+						supply(() -> {
+							MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+							multiSelectFilter.setEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setMultiSelect(multiSelectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().multiSelect("property").isNotEmpty(), //
+						supply(() -> {
+							MultiSelectFilter multiSelectFilter = new MultiSelectFilter();
+							multiSelectFilter.setNotEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setMultiSelect(multiSelectFilter);
+							return propertyFilter;
+						})));
+	}
+
+	static Stream<Arguments> numberFilters() {
+		return Stream.of( //
+				arguments( //
+						where().number("property").isEqualTo(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setEquals(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isNotEqualTo(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setDoesNotEqual(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isGreaterThan(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setGreaterThan(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isGreaterThanOrEqualTo(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setGreaterThanOrEqualTo(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isLessThan(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setLessThan(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isLessThanOrEqualTo(42), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setLessThanOrEqualTo(42);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isEmpty(), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().number("property").isNotEmpty(), //
+						supply(() -> {
+							NumberFilter numberFilter = new NumberFilter();
+							numberFilter.setNotEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setNumber(numberFilter);
+							return propertyFilter;
+						})));
+	}
+
+	static Stream<Arguments> selectFilters() {
+		return Stream.of( //
+				arguments( //
+						where().select("property").isEqualTo("value"), //
+						supply(() -> {
+							SelectFilter selectFilter = new SelectFilter();
+							selectFilter.setEquals("value");
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setSelect(selectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().select("property").isNotEqualTo("value"), //
+						supply(() -> {
+							SelectFilter selectFilter = new SelectFilter();
+							selectFilter.setDoesNotEqual("value");
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setSelect(selectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().select("property").isEmpty(), //
+						supply(() -> {
+							SelectFilter selectFilter = new SelectFilter();
+							selectFilter.setEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setSelect(selectFilter);
+							return propertyFilter;
+						})),
+				arguments( //
+						where().select("property").isNotEmpty(), //
+						supply(() -> {
+							SelectFilter selectFilter = new SelectFilter();
+							selectFilter.setNotEmpty(true);
+							PropertyFilter propertyFilter = new PropertyFilter("property");
+							propertyFilter.setSelect(selectFilter);
+							return propertyFilter;
+						})));
+	}
+
+	static Stream<Arguments> compoundFilters() {
+		return Stream.of(andFilters(), orFilters()).flatMap(identity());
+	}
+
+	static Stream<Arguments> andFilters() {
+		return Stream.of( //
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+							.and().select("another").isNotEmpty(), //
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setAnd(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})),
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+							.and(where().select("another").isNotEmpty()),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setAnd(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})),
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+							.and(where().select("another").isNotEmpty())
+							.and().checkbox("one-more").isNotEqualTo(true)
+							.and(where().select("another-more").isEmpty()),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setAnd(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setDoesNotEqual(true);
+										PropertyFilter propertyFilter = new PropertyFilter("one-more");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another-more");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})));
+	}
+
+	static Stream<Arguments> orFilters() {
+		return Stream.of( //
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+							.or().select("another").isNotEmpty(),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setOr(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})),
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+							.or(where().select("another").isNotEmpty()),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setOr(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})),
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(false)
+								.or(where().select("another").isNotEmpty())
+								.or().checkbox("one-more").isNotEqualTo(true)
+								.or(where().select("another-more").isEmpty()),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setOr(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(false);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setNotEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setDoesNotEqual(true);
+										PropertyFilter propertyFilter = new PropertyFilter("one-more");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("another-more");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})));
+	}
+
+	static Stream<Arguments> nestedFilters() {
+		return Stream.of( //
+				arguments( // @formatter:off
+						where(where().checkbox("active").isEqualTo(true)),
+						// @formatter:on
+						supply(() -> {
+							CheckboxFilter checkboxFilter = new CheckboxFilter();
+							checkboxFilter.setEquals(true);
+							PropertyFilter propertyFilter = new PropertyFilter("active");
+							propertyFilter.setCheckbox(checkboxFilter);
+							return propertyFilter;
+						})),
+				arguments( // @formatter:off
+						where().checkbox("active").isEqualTo(true)
+							.and(where().select("another").isEmpty().or().select("another").isEqualTo("value")),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setAnd(List.of( //
+									supply(() -> {
+										CheckboxFilter checkboxFilter = new CheckboxFilter();
+										checkboxFilter.setEquals(true);
+										PropertyFilter propertyFilter = new PropertyFilter("active");
+										propertyFilter.setCheckbox(checkboxFilter);
+										return propertyFilter;
+									}), //
+									supply(() -> {
+										CompoundFilter innerFilter = new CompoundFilter();
+
+										innerFilter.setOr(List.of( //
+												supply(() -> {
+													SelectFilter selectFilter = new SelectFilter();
+													selectFilter.setEmpty(true);
+													PropertyFilter propertyFilter = new PropertyFilter("another");
+													propertyFilter.setSelect(selectFilter);
+													return propertyFilter;
+												}), //
+												supply(() -> {
+													SelectFilter selectFilter = new SelectFilter();
+													selectFilter.setEquals("value");
+													PropertyFilter propertyFilter = new PropertyFilter("another");
+													propertyFilter.setSelect(selectFilter);
+													return propertyFilter;
+												})));
+
+										return innerFilter;
+									})));
+
+							return compoundFilter;
+						})),
+				arguments( // @formatter:off
+						where(where().checkbox("active").isEqualTo(false)
+								.or().select("another").isNotEmpty())
+							.and().select("one-more").isEmpty(),
+						// @formatter:on
+						supply(() -> {
+							CompoundFilter compoundFilter = new CompoundFilter();
+
+							compoundFilter.setAnd(List.of( //
+									supply(() -> {
+										CompoundFilter innerFilter = new CompoundFilter();
+
+										innerFilter.setOr(List.of( //
+												supply(() -> {
+													CheckboxFilter checkboxFilter = new CheckboxFilter();
+													checkboxFilter.setEquals(false);
+													PropertyFilter propertyFilter = new PropertyFilter("active");
+													propertyFilter.setCheckbox(checkboxFilter);
+													return propertyFilter;
+												}), //
+												supply(() -> {
+													SelectFilter selectFilter = new SelectFilter();
+													selectFilter.setNotEmpty(true);
+													PropertyFilter propertyFilter = new PropertyFilter("another");
+													propertyFilter.setSelect(selectFilter);
+													return propertyFilter;
+												})));
+
+										return innerFilter;
+									}), //
+									supply(() -> {
+										SelectFilter selectFilter = new SelectFilter();
+										selectFilter.setEmpty(true);
+										PropertyFilter propertyFilter = new PropertyFilter("one-more");
+										propertyFilter.setSelect(selectFilter);
+										return propertyFilter;
+									})));
+
+							return compoundFilter;
+						})));
+	}
+
+	private static <T> T supply(Supplier<T> supplier) {
+		return supplier.get();
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/NotionJvmSdkTests.java

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import com.tngtech.archunit.base.DescribedPredicate;
+import com.tngtech.archunit.core.domain.JavaClass;
+import com.tngtech.archunit.core.domain.JavaClasses;
+import com.tngtech.archunit.junit.AnalyzeClasses;
+import com.tngtech.archunit.junit.ArchTest;
+import com.tngtech.archunit.lang.ArchRule;
+
+import static com.tngtech.archunit.base.DescribedPredicate.anyElementThat;
+import static com.tngtech.archunit.core.domain.JavaClass.Predicates.resideInAPackage;
+import static com.tngtech.archunit.lang.syntax.ArchRuleDefinition.methods;
+
+/**
+ * @author Stefano Cordio
+ */
+@AnalyzeClasses(packagesOf = NotionDatabaseItemReader.class)
+class NotionJvmSdkTests {
+
+	private static final DescribedPredicate<JavaClass> RESIDE_IN_NOTION_JVM_SDK_PACKAGE = //
+			resideInAPackage("notion.api..");
+
+	@ArchTest
+	void library_types_should_not_be_exposed(JavaClasses classes) {
+		// @formatter:off
+		ArchRule rule = methods()
+				.that().arePublic().or().areProtected()
+				.should().notHaveRawReturnType(RESIDE_IN_NOTION_JVM_SDK_PACKAGE)
+				.andShould().notHaveRawParameterTypes(anyElementThat(RESIDE_IN_NOTION_JVM_SDK_PACKAGE));
+		// @formatter:on
+		rule.check(classes);
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/SortTests.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion;
+
+import notion.api.v1.model.databases.query.sort.QuerySort;
+import notion.api.v1.model.databases.query.sort.QuerySortDirection;
+import notion.api.v1.model.databases.query.sort.QuerySortTimestamp;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.stream.Stream;
+
+import static org.springframework.batch.extensions.notion.Sort.Direction.ASCENDING;
+import static org.springframework.batch.extensions.notion.Sort.Direction.DESCENDING;
+import static org.springframework.batch.extensions.notion.Sort.Timestamp.CREATED_TIME;
+import static org.springframework.batch.extensions.notion.Sort.Timestamp.LAST_EDITED_TIME;
+import static notion.api.v1.model.databases.query.sort.QuerySortDirection.Ascending;
+import static notion.api.v1.model.databases.query.sort.QuerySortDirection.Descending;
+import static notion.api.v1.model.databases.query.sort.QuerySortTimestamp.CreatedTime;
+import static notion.api.v1.model.databases.query.sort.QuerySortTimestamp.LastEditedTime;
+import static org.assertj.core.api.BDDAssertions.from;
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+/**
+ * @author Stefano Cordio
+ */
+class SortTests {
+
+	@ParameterizedTest
+	@MethodSource
+	void toQuerySort(Sort underTest, String property, QuerySortTimestamp timestamp, QuerySortDirection direction) {
+		// WHEN
+		QuerySort result = underTest.toQuerySort();
+		// THEN
+		then(result) //
+			.returns(direction, from(QuerySort::getDirection))
+			.returns(property, from(QuerySort::getProperty))
+			.returns(timestamp, from(QuerySort::getTimestamp));
+	}
+
+	static Stream<Arguments> toQuerySort() {
+		return Stream.of( //
+				arguments(Sort.by("property"), "property", null, Ascending),
+				arguments(Sort.by("property", ASCENDING), "property", null, Ascending),
+				arguments(Sort.by("property", DESCENDING), "property", null, Descending),
+				arguments(Sort.by(CREATED_TIME), null, CreatedTime, Ascending),
+				arguments(Sort.by(CREATED_TIME, ASCENDING), null, CreatedTime, Ascending),
+				arguments(Sort.by(CREATED_TIME, DESCENDING), null, CreatedTime, Descending),
+				arguments(Sort.by(LAST_EDITED_TIME), null, LastEditedTime, Ascending),
+				arguments(Sort.by(LAST_EDITED_TIME, ASCENDING), null, LastEditedTime, Ascending),
+				arguments(Sort.by(LAST_EDITED_TIME, DESCENDING), null, LastEditedTime, Descending));
+	}
+
+	@ParameterizedTest
+	@MethodSource
+	void testToString(Sort underTest, String expected) {
+		// WHEN
+		String result = underTest.toString();
+		// THEN
+		then(result).isEqualTo(expected);
+	}
+
+	static Stream<Arguments> testToString() {
+		return Stream.of( //
+				arguments(Sort.by("property"), "property: ASCENDING"),
+				arguments(Sort.by("property", ASCENDING), "property: ASCENDING"),
+				arguments(Sort.by("property", DESCENDING), "property: DESCENDING"),
+				arguments(Sort.by(CREATED_TIME), "CREATED_TIME: ASCENDING"),
+				arguments(Sort.by(CREATED_TIME, ASCENDING), "CREATED_TIME: ASCENDING"),
+				arguments(Sort.by(CREATED_TIME, DESCENDING), "CREATED_TIME: DESCENDING"),
+				arguments(Sort.by(LAST_EDITED_TIME), "LAST_EDITED_TIME: ASCENDING"),
+				arguments(Sort.by(LAST_EDITED_TIME, ASCENDING), "LAST_EDITED_TIME: ASCENDING"),
+				arguments(Sort.by(LAST_EDITED_TIME, DESCENDING), "LAST_EDITED_TIME: DESCENDING"));
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/IntegrationTest.java

@@ -0,0 +1,38 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it;
+
+import org.springframework.batch.test.context.SpringBatchTest;
+import org.springframework.boot.test.context.SpringBootTest;
+import org.wiremock.spring.EnableWireMock;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.TYPE;
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * @author Stefano Cordio
+ */
+@Target(TYPE)
+@Retention(RUNTIME)
+@SpringBootTest(properties = "spring.batch.job.enabled=false")
+@SpringBatchTest
+@EnableWireMock
+public @interface IntegrationTest {
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/RequestBodies.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it;
+
+import org.springframework.batch.extensions.notion.Sort.Direction;
+import org.springframework.batch.extensions.notion.Sort.Timestamp;
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import java.util.UUID;
+
+/**
+ * @author Stefano Cordio
+ */
+public class RequestBodies {
+
+	public static String queryRequest(int pageSize, JSONObject... sorts) {
+		return queryRequest(null, pageSize, sorts);
+	}
+
+	public static String queryRequest(UUID startCursor, int pageSize, JSONObject... sorts) {
+		try {
+			JSONObject jsonObject = new JSONObject();
+
+			if (sorts.length > 0) {
+				jsonObject.put("sorts", new JSONArray(sorts));
+			}
+
+			return jsonObject //
+				.put("page_size", pageSize)
+				.putOpt("start_cursor", startCursor != null ? startCursor.toString() : null)
+				.toString();
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	public static JSONObject sortByProperty(String property, Direction direction) {
+		try {
+			return new JSONObject() //
+				.put("property", property)
+				.put("direction", direction.name().toLowerCase());
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	public static JSONObject sortByTimestamp(String property, Timestamp timestamp) {
+		try {
+			return new JSONObject() //
+				.put("property", property)
+				.put("timestamp", timestamp.name().toLowerCase());
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/RequestHeaders.java

@@ -0,0 +1,27 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it;
+
+/**
+ * @author Stefano Cordio
+ */
+public class RequestHeaders {
+
+	public static final String NOTION_VERSION = "Notion-Version";
+
+	public static final String NOTION_VERSION_VALUE = "2022-06-28";
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/ResponseBodies.java

@@ -0,0 +1,129 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it;
+
+import org.json.JSONArray;
+import org.json.JSONException;
+import org.json.JSONObject;
+
+import java.time.Instant;
+import java.util.Map;
+import java.util.UUID;
+
+import static java.util.UUID.randomUUID;
+
+/**
+ * @author Stefano Cordio
+ */
+public class ResponseBodies {
+
+	public static String queryResponse(JSONObject... results) {
+		return queryResponse(null, results);
+	}
+
+	public static String queryResponse(UUID nextCursor, JSONObject... results) {
+		try {
+			return new JSONObject() //
+				.put("object", "list")
+				.put("results", new JSONArray(results))
+				.put("next_cursor", nextCursor != null ? nextCursor.toString() : null)
+				.put("has_more", nextCursor != null)
+				.put("type", "page")
+				.put("page", new JSONObject())
+				.toString();
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	public static JSONObject result(UUID id, UUID databaseId, Map<?, ?> properties) {
+		try {
+			Instant now = Instant.now();
+
+			return new JSONObject() //
+				.put("object", "page")
+				.put("id", id.toString())
+				.put("created_time", now.toString())
+				.put("last_edited_time", now.toString())
+				.put("created_by", new JSONObject())
+				.put("last_edited_by", new JSONObject())
+				.put("parent", new JSONObject() //
+					.put("type", "database_id")
+					.put("database_id", databaseId.toString()))
+				.put("archived", false)
+				.put("properties", new JSONObject(properties))
+				.put("url", "https://www.notion.so/" + randomUUID().toString().replace("-", ""));
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	public static JSONObject title(String value) {
+		try {
+			JSONArray jsonArray = new JSONArray();
+
+			if (value != null) {
+				jsonArray.put(new JSONObject() //
+					.put("type", "text")
+					.put("text", new JSONObject() //
+						.put("content", value))
+					.put("annotations", new JSONObject() //
+						.put("bold", false)
+						.put("italic", false)
+						.put("strikethrough", false)
+						.put("underline", false)
+						.put("code", false)
+						.put("color", "default"))
+					.put("plain_text", value));
+			}
+
+			return new JSONObject() //
+				.put("id", "title")
+				.put("type", "title")
+				.put("title", jsonArray);
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+	public static JSONObject richText(String value) {
+		try {
+			return new JSONObject() //
+				.put("id", "JV%3B%3F")
+				.put("type", "rich_text")
+				.put("rich_text", new JSONArray() //
+					.put(new JSONObject() //
+						.put("type", "text")
+						.put("text", new JSONObject() //
+							.put("content", value))
+						.put("annotations", new JSONObject() //
+							.put("bold", false)
+							.put("italic", false)
+							.put("strikethrough", false)
+							.put("underline", false)
+							.put("code", false)
+							.put("color", "default"))
+						.put("plain_text", value)));
+		}
+		catch (JSONException e) {
+			throw new RuntimeException(e);
+		}
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/pagination/MultiplePagesDescendingTests.java

@@ -0,0 +1,166 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it.pagination;
+
+import org.springframework.batch.extensions.notion.NotionDatabaseItemReader;
+import org.springframework.batch.extensions.notion.Sort;
+import org.springframework.batch.extensions.notion.it.IntegrationTest;
+import org.springframework.batch.extensions.notion.mapping.RecordPropertyMapper;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.builder.JobBuilder;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.item.support.ListItemWriter;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import java.util.Map;
+import java.util.UUID;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalToJson;
+import static com.github.tomakehurst.wiremock.client.WireMock.givenThat;
+import static com.github.tomakehurst.wiremock.client.WireMock.matching;
+import static com.github.tomakehurst.wiremock.client.WireMock.okJson;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static org.springframework.batch.extensions.notion.Sort.Direction.DESCENDING;
+import static org.springframework.batch.extensions.notion.it.RequestBodies.queryRequest;
+import static org.springframework.batch.extensions.notion.it.RequestBodies.sortByProperty;
+import static org.springframework.batch.extensions.notion.it.RequestHeaders.NOTION_VERSION;
+import static org.springframework.batch.extensions.notion.it.RequestHeaders.NOTION_VERSION_VALUE;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.queryResponse;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.result;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.richText;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.title;
+import static java.util.UUID.randomUUID;
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.assertj.core.api.InstanceOfAssertFactories.LIST;
+import static org.springframework.batch.core.ExitStatus.COMPLETED;
+import static wiremock.com.google.common.net.HttpHeaders.AUTHORIZATION;
+import static wiremock.com.google.common.net.HttpHeaders.CONTENT_TYPE;
+
+/**
+ * @author Stefano Cordio
+ */
+@IntegrationTest
+class MultiplePagesDescendingTests {
+
+	private static final UUID DATABASE_ID = randomUUID();
+
+	private static final int PAGE_SIZE = 2;
+
+	@Autowired
+	JobLauncherTestUtils launcher;
+
+	@Autowired
+	ListItemWriter<PaginatedDescendingJob.Item> itemWriter;
+
+	@Test
+	void should_succeed() throws Exception {
+		// GIVEN
+		UUID thirdResultId = randomUUID();
+
+		JSONObject firstResult = result(randomUUID(), DATABASE_ID,
+				Map.of("Name", title("Name string"), "Value", richText("123456")));
+		JSONObject secondResult = result(randomUUID(), DATABASE_ID,
+				Map.of("Name", title("Another name string"), "Value", richText("0987654321")));
+		JSONObject thirdResult = result(thirdResultId, DATABASE_ID,
+				Map.of("Name", title(""), "Value", richText("abc-1234")));
+
+		givenThat(post("/databases/%s/query".formatted(DATABASE_ID)) //
+			.withHeader(AUTHORIZATION, matching("Bearer .+"))
+			.withHeader(CONTENT_TYPE, containing("application/json"))
+			.withHeader(NOTION_VERSION, equalTo(NOTION_VERSION_VALUE))
+			.withRequestBody(equalToJson(queryRequest(PAGE_SIZE, sortByProperty("Name", DESCENDING))))
+			.willReturn(okJson(queryResponse(thirdResultId, firstResult, secondResult))));
+
+		givenThat(post("/databases/%s/query".formatted(DATABASE_ID)) //
+			.withHeader(AUTHORIZATION, matching("Bearer .+"))
+			.withHeader(CONTENT_TYPE, containing("application/json"))
+			.withHeader(NOTION_VERSION, equalTo(NOTION_VERSION_VALUE))
+			.withRequestBody(equalToJson(queryRequest(thirdResultId, PAGE_SIZE, sortByProperty("Name", DESCENDING))))
+			.willReturn(okJson(queryResponse(thirdResult))));
+
+		// WHEN
+		JobExecution jobExecution = launcher.launchJob();
+
+		// THEN
+		then(jobExecution.getExitStatus()).isEqualTo(COMPLETED);
+
+		then(itemWriter.getWrittenItems()).asInstanceOf(LIST)
+			.containsExactly( //
+					new PaginatedDescendingJob.Item("Name string", "123456"), //
+					new PaginatedDescendingJob.Item("Another name string", "0987654321"), //
+					new PaginatedDescendingJob.Item("", "abc-1234"));
+	}
+
+	@SpringBootApplication
+	static class PaginatedDescendingJob {
+
+		@Value("${wiremock.server.baseUrl}")
+		private String wiremockBaseUrl;
+
+		@Bean
+		Job job(JobRepository jobRepository, Step step) {
+			return new JobBuilder("TEST-JOB", jobRepository).start(step).build();
+		}
+
+		@Bean
+		Step step(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
+			return new StepBuilder("TEST-STEP", jobRepository) //
+				.<Item, Item>chunk(PAGE_SIZE, transactionManager) //
+				.reader(itemReader()) //
+				.writer(itemWriter()) //
+				.build();
+		}
+
+		@Bean
+		NotionDatabaseItemReader<Item> itemReader() {
+			NotionDatabaseItemReader<Item> reader = new NotionDatabaseItemReader<>();
+
+			reader.setSaveState(false);
+
+			reader.setToken("token");
+			reader.setBaseUrl(wiremockBaseUrl);
+			reader.setDatabaseId(DATABASE_ID.toString());
+
+			reader.setPageSize(PAGE_SIZE);
+			reader.setSorts(Sort.by("Name", DESCENDING));
+			reader.setPropertyMapper(new RecordPropertyMapper<>());
+
+			return reader;
+		}
+
+		@Bean
+		ListItemWriter<Item> itemWriter() {
+			return new ListItemWriter<>();
+		}
+
+		record Item(String name, String value) {
+		}
+
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/it/pagination/MultiplePagesTests.java

@@ -0,0 +1,162 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.it.pagination;
+
+import org.springframework.batch.extensions.notion.NotionDatabaseItemReader;
+import org.springframework.batch.extensions.notion.it.IntegrationTest;
+import org.springframework.batch.extensions.notion.mapping.RecordPropertyMapper;
+import org.json.JSONObject;
+import org.junit.jupiter.api.Test;
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobExecution;
+import org.springframework.batch.core.Step;
+import org.springframework.batch.core.job.builder.JobBuilder;
+import org.springframework.batch.core.repository.JobRepository;
+import org.springframework.batch.core.step.builder.StepBuilder;
+import org.springframework.batch.item.support.ListItemWriter;
+import org.springframework.batch.test.JobLauncherTestUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Bean;
+import org.springframework.transaction.PlatformTransactionManager;
+
+import java.util.Map;
+import java.util.UUID;
+
+import static com.github.tomakehurst.wiremock.client.WireMock.containing;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalTo;
+import static com.github.tomakehurst.wiremock.client.WireMock.equalToJson;
+import static com.github.tomakehurst.wiremock.client.WireMock.givenThat;
+import static com.github.tomakehurst.wiremock.client.WireMock.matching;
+import static com.github.tomakehurst.wiremock.client.WireMock.okJson;
+import static com.github.tomakehurst.wiremock.client.WireMock.post;
+import static org.springframework.batch.extensions.notion.it.RequestBodies.queryRequest;
+import static org.springframework.batch.extensions.notion.it.RequestHeaders.NOTION_VERSION;
+import static org.springframework.batch.extensions.notion.it.RequestHeaders.NOTION_VERSION_VALUE;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.queryResponse;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.result;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.richText;
+import static org.springframework.batch.extensions.notion.it.ResponseBodies.title;
+import static java.util.UUID.randomUUID;
+import static org.assertj.core.api.BDDAssertions.then;
+import static org.assertj.core.api.InstanceOfAssertFactories.LIST;
+import static org.springframework.batch.core.ExitStatus.COMPLETED;
+import static wiremock.com.google.common.net.HttpHeaders.AUTHORIZATION;
+import static wiremock.com.google.common.net.HttpHeaders.CONTENT_TYPE;
+
+/**
+ * @author Stefano Cordio
+ */
+@IntegrationTest
+class MultiplePagesTests {
+
+	private static final UUID DATABASE_ID = randomUUID();
+
+	private static final int PAGE_SIZE = 2;
+
+	@Autowired
+	JobLauncherTestUtils launcher;
+
+	@Autowired
+	ListItemWriter<PaginatedJob.Item> itemWriter;
+
+	@Test
+	void should_succeed() throws Exception {
+		// GIVEN
+		UUID thirdResultId = randomUUID();
+
+		JSONObject firstResult = result(randomUUID(), DATABASE_ID,
+				Map.of("Name", title("Another name string"), "Value", richText("0987654321")));
+		JSONObject secondResult = result(randomUUID(), DATABASE_ID,
+				Map.of("Name", title("Name string"), "Value", richText("123456")));
+		JSONObject thirdResult = result(thirdResultId, DATABASE_ID,
+				Map.of("Name", title(""), "Value", richText("abc-1234")));
+
+		givenThat(post("/databases/%s/query".formatted(DATABASE_ID)) //
+			.withHeader(AUTHORIZATION, matching("Bearer .+"))
+			.withHeader(CONTENT_TYPE, containing("application/json"))
+			.withHeader(NOTION_VERSION, equalTo(NOTION_VERSION_VALUE))
+			.withRequestBody(equalToJson(queryRequest(PAGE_SIZE)))
+			.willReturn(okJson(queryResponse(thirdResultId, firstResult, secondResult))));
+
+		givenThat(post("/databases/%s/query".formatted(DATABASE_ID)) //
+			.withHeader(AUTHORIZATION, matching("Bearer .+"))
+			.withHeader(CONTENT_TYPE, containing("application/json"))
+			.withHeader(NOTION_VERSION, equalTo(NOTION_VERSION_VALUE))
+			.withRequestBody(equalToJson(queryRequest(thirdResultId, PAGE_SIZE)))
+			.willReturn(okJson(queryResponse(thirdResult))));
+
+		// WHEN
+		JobExecution jobExecution = launcher.launchJob();
+
+		// THEN
+		then(jobExecution.getExitStatus()).isEqualTo(COMPLETED);
+
+		then(itemWriter.getWrittenItems()).asInstanceOf(LIST)
+			.containsExactly( //
+					new PaginatedJob.Item("Another name string", "0987654321"), //
+					new PaginatedJob.Item("Name string", "123456"), //
+					new PaginatedJob.Item("", "abc-1234"));
+	}
+
+	@SpringBootApplication
+	static class PaginatedJob {
+
+		@Value("${wiremock.server.baseUrl}")
+		private String wiremockBaseUrl;
+
+		@Bean
+		Job job(JobRepository jobRepository, Step step) {
+			return new JobBuilder("TEST-JOB", jobRepository).start(step).build();
+		}
+
+		@Bean
+		Step step(JobRepository jobRepository, PlatformTransactionManager transactionManager) {
+			return new StepBuilder("TEST-STEP", jobRepository) //
+				.<Item, Item>chunk(PAGE_SIZE, transactionManager) //
+				.reader(itemReader()) //
+				.writer(itemWriter()) //
+				.build();
+		}
+
+		@Bean
+		NotionDatabaseItemReader<Item> itemReader() {
+			NotionDatabaseItemReader<Item> reader = new NotionDatabaseItemReader<>();
+
+			reader.setSaveState(false);
+
+			reader.setToken("token");
+			reader.setBaseUrl(wiremockBaseUrl);
+			reader.setDatabaseId(DATABASE_ID.toString());
+
+			reader.setPageSize(PAGE_SIZE);
+			reader.setPropertyMapper(new RecordPropertyMapper<>());
+
+			return reader;
+		}
+
+		@Bean
+		ListItemWriter<Item> itemWriter() {
+			return new ListItemWriter<>();
+		}
+
+		record Item(String name, String value) {
+		}
+
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/mapping/BeanWrapperPropertyMapperTests.java

@@ -0,0 +1,109 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.batch.extensions.notion.mapping.TestData.AllPropertiesSource;
+import org.springframework.batch.extensions.notion.mapping.TestData.PartialPropertiesSource;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+
+import java.util.Map;
+
+import static org.assertj.core.api.BDDAssertions.catchThrowable;
+import static org.assertj.core.api.BDDAssertions.from;
+import static org.assertj.core.api.BDDAssertions.then;
+
+/**
+ * @author Stefano Cordio
+ */
+class BeanWrapperPropertyMapperTests {
+
+	@ParameterizedTest
+	@AllPropertiesSource
+	void should_map_all_properties(Map<String, String> properties) {
+		// GIVEN
+		PropertyMapper<TestBean> underTest = new BeanWrapperPropertyMapper<>(TestBean.class);
+		// WHEN
+		TestBean result = underTest.map(properties);
+		// THEN
+		then(result) //
+			.returns("Value1", from(TestBean::getField1)) //
+			.returns("Value2", from(TestBean::getField2));
+	}
+
+	@ParameterizedTest
+	@PartialPropertiesSource
+	void should_map_partial_properties(Map<String, String> properties) {
+		// GIVEN
+		PropertyMapper<TestBean> underTest = new BeanWrapperPropertyMapper<>(TestBean.class);
+		// WHEN
+		TestBean result = underTest.map(properties);
+		// THEN
+		then(result) //
+			.returns("Value1", from(TestBean::getField1)) //
+			.returns(null, from(TestBean::getField2));
+	}
+
+	@ParameterizedTest
+	@AllPropertiesSource
+	void should_map_all_properties_without_type_parameter(Map<String, String> properties) {
+		// GIVEN
+		PropertyMapper<TestBean> underTest = new BeanWrapperPropertyMapper<>();
+		// WHEN
+		TestBean result = underTest.map(properties);
+		// THEN
+		then(result) //
+			.returns("Value1", from(TestBean::getField1)) //
+			.returns("Value2", from(TestBean::getField2));
+	}
+
+	@Test
+	void should_fail_with_vararg_constructor_parameter() {
+		// WHEN
+		Throwable thrown = catchThrowable(() -> new BeanWrapperPropertyMapper<>(new TestBean()));
+		// THEN
+		then(thrown) //
+			.isInstanceOf(IllegalArgumentException.class)
+			.hasMessage("Please don't pass any values here. The type will be detected automagically.");
+	}
+
+	private static class TestBean {
+
+		private String field1;
+
+		private String field2;
+
+		public String getField1() {
+			return field1;
+		}
+
+		@SuppressWarnings("unused")
+		public void setField1(String field1) {
+			this.field1 = field1;
+		}
+
+		public String getField2() {
+			return field2;
+		}
+
+		@SuppressWarnings("unused")
+		public void setField2(String field2) {
+			this.field2 = field2;
+		}
+
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/mapping/ConstructorPropertyMapperTests.java

@@ -0,0 +1,122 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.batch.extensions.notion.mapping.TestData.AllPropertiesSource;
+import org.springframework.batch.extensions.notion.mapping.TestData.PartialPropertiesSource;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+
+import java.util.Map;
+
+import static org.assertj.core.api.BDDAssertions.catchThrowable;
+import static org.assertj.core.api.BDDAssertions.from;
+import static org.assertj.core.api.BDDAssertions.then;
+
+/**
+ * @author Stefano Cordio
+ */
+class ConstructorPropertyMapperTests {
+
+	@Nested
+	class using_record_without_additional_constructors {
+
+		private record TestRecord(String field1, String field2) {
+		}
+
+		@ParameterizedTest
+		@AllPropertiesSource
+		void should_map_all_properties(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new ConstructorPropertyMapper<>(TestRecord.class);
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns("Value2", from(TestRecord::field2));
+		}
+
+		@ParameterizedTest
+		@AllPropertiesSource
+		void should_map_all_properties_without_type_parameter(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new ConstructorPropertyMapper<>();
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns("Value2", from(TestRecord::field2));
+		}
+
+		@Test
+		void should_fail_with_vararg_constructor_parameter() {
+			// WHEN
+			Throwable thrown = catchThrowable(() -> new ConstructorPropertyMapper<>(new TestRecord("value", "value")));
+			// THEN
+			then(thrown) //
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessage("Please don't pass any values here. The type will be detected automagically.");
+		}
+
+		@ParameterizedTest
+		@PartialPropertiesSource
+		void should_map_partial_properties(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new ConstructorPropertyMapper<>(TestRecord.class);
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns(null, from(TestRecord::field2));
+		}
+
+	}
+
+	@Nested
+	class using_record_with_additional_constructors {
+
+		private record TestRecord(String field1, String field2) {
+
+			@SuppressWarnings("unused")
+			private TestRecord() {
+				this(null, null);
+			}
+
+			@SuppressWarnings("unused")
+			private TestRecord(String field1) {
+				this(field1, null);
+			}
+
+		}
+
+		@Test
+		void should_fail() {
+			// WHEN
+			Throwable thrown = catchThrowable(() -> new ConstructorPropertyMapper<>(TestRecord.class));
+			// THEN
+			then(thrown).isInstanceOf(IllegalArgumentException.class) //
+				.cause() //
+				.isInstanceOf(NoSuchMethodException.class)
+				.hasMessageStartingWith("Multiple constructors available: ");
+		}
+
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/mapping/RecordPropertyMapperTests.java

@@ -0,0 +1,124 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.springframework.batch.extensions.notion.mapping.TestData.AllPropertiesSource;
+import org.springframework.batch.extensions.notion.mapping.TestData.PartialPropertiesSource;
+import org.junit.jupiter.api.Nested;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+
+import java.util.Map;
+
+import static org.assertj.core.api.BDDAssertions.catchThrowable;
+import static org.assertj.core.api.BDDAssertions.from;
+import static org.assertj.core.api.BDDAssertions.then;
+
+/**
+ * @author Stefano Cordio
+ */
+class RecordPropertyMapperTests {
+
+	@Nested
+	class using_record_without_additional_constructors {
+
+		private record TestRecord(String field1, String field2) {
+		}
+
+		@ParameterizedTest
+		@AllPropertiesSource
+		void should_map_all_properties(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new RecordPropertyMapper<>(TestRecord.class);
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns("Value2", from(TestRecord::field2));
+		}
+
+		@ParameterizedTest
+		@AllPropertiesSource
+		void should_map_all_properties_without_type_parameter(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new RecordPropertyMapper<>();
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns("Value2", from(TestRecord::field2));
+		}
+
+		@Test
+		void should_fail_with_vararg_constructor_parameter() {
+			// WHEN
+			Throwable thrown = catchThrowable(() -> new RecordPropertyMapper<>(new TestRecord("value", "value")));
+			// THEN
+			then(thrown) //
+				.isInstanceOf(IllegalArgumentException.class)
+				.hasMessage("Please don't pass any values here. The type will be detected automagically.");
+		}
+
+		@ParameterizedTest
+		@PartialPropertiesSource
+		void should_map_partial_properties(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new RecordPropertyMapper<>(TestRecord.class);
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns(null, from(TestRecord::field2));
+		}
+
+	}
+
+	@Nested
+	class using_record_with_additional_constructors {
+
+		private record TestRecord(String field1, String field2) {
+
+			@SuppressWarnings("unused")
+			private TestRecord() {
+				this(null, null);
+			}
+
+			@SuppressWarnings("unused")
+			private TestRecord(String field1) {
+				this(field1, null);
+			}
+
+		}
+
+		@ParameterizedTest
+		@AllPropertiesSource
+		void should_map_all_properties(Map<String, String> properties) {
+			// GIVEN
+			PropertyMapper<TestRecord> underTest = new RecordPropertyMapper<>(TestRecord.class);
+			// WHEN
+			TestRecord result = underTest.map(properties);
+			// THEN
+			then(result) //
+				.returns("Value1", from(TestRecord::field1)) //
+				.returns("Value2", from(TestRecord::field2));
+		}
+
+	}
+
+}

spring-batch-notion/src/test/java/org/springframework/batch/extensions/notion/mapping/TestData.java

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2002-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.batch.extensions.notion.mapping;
+
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.lang.annotation.Retention;
+import java.util.Map;
+import java.util.stream.Stream;
+
+import static java.lang.annotation.RetentionPolicy.RUNTIME;
+
+/**
+ * @author Stefano Cordio
+ */
+class TestData {
+
+	@Retention(RUNTIME)
+	@MethodSource("org.springframework.batch.extensions.notion.mapping.TestData#all_properties")
+	@interface AllPropertiesSource {
+
+	}
+
+	static Stream<Map<String, String>> all_properties() {
+		return Stream.of(
+				// FIXME not working with BeanWrapperPropertyMapper
+				// Map.of("FIELD1", "Value1", "FIELD2", "Value2"), //
+				Map.of("Field1", "Value1", "Field2", "Value2"), //
+				Map.of("field1", "Value1", "field2", "Value2"));
+	}
+
+	@Retention(RUNTIME)
+	@MethodSource("org.springframework.batch.extensions.notion.mapping.TestData#partial_properties")
+	@interface PartialPropertiesSource {
+
+	}
+
+	static Stream<Map<String, String>> partial_properties() {
+		return Stream.of(
+				// FIXME not working with BeanWrapperPropertyMapper
+				// Map.of("FIELD1", "Value1"), //
+				Map.of("Field1", "Value1"), //
+				Map.of("field1", "Value1"));
+	}
+
+}

spring-batch-notion/src/test/resources/junit-platform.properties

@@ -0,0 +1 @@
+junit.jupiter.displayname.generator.default=org.junit.jupiter.api.DisplayNameGenerator$ReplaceUnderscores