Move Transaction code to stac-fastapi.extension

[vincentsarago]

There are code in stac_fastapi.extension for transaction client and models which IMO should be in the extension sub-module

stac-fastapi/stac_fastapi/types/stac_fastapi/types/core.py

Lines 39 to 264 in 20ae9cf

	@attr.s # type:ignore
	class BaseTransactionsClient(abc.ABC):
	"""Defines a pattern for implementing the STAC API Transaction Extension."""
	@abc.abstractmethod
	def create_item(
	self,
	collection_id: str,
	item: Union[Item, ItemCollection],
	**kwargs,
	) -> Optional[Union[stac.Item, Response, None]]:
	"""Create a new item.
	Called with `POST /collections/{collection_id}/items`.
	Args:
	item: the item or item collection
	collection_id: the id of the collection from the resource path
	Returns:
	The item that was created or None if item collection.
	"""
	...
	@abc.abstractmethod
	def update_item(
	self, collection_id: str, item_id: str, item: Item, **kwargs
	) -> Optional[Union[stac.Item, Response]]:
	"""Perform a complete update on an existing item.
	Called with `PUT /collections/{collection_id}/items`. It is expected
	that this item already exists. The update should do a diff against the
	saved item and perform any necessary updates. Partial updates are not
	supported by the transactions extension.
	Args:
	item: the item (must be complete)
	collection_id: the id of the collection from the resource path
	Returns:
	The updated item.
	"""
	...
	@abc.abstractmethod
	def delete_item(
	self, item_id: str, collection_id: str, **kwargs
	) -> Optional[Union[stac.Item, Response]]:
	"""Delete an item from a collection.
	Called with `DELETE /collections/{collection_id}/items/{item_id}`
	Args:
	item_id: id of the item.
	collection_id: id of the collection.
	Returns:
	The deleted item.
	"""
	...
	@abc.abstractmethod
	def create_collection(
	self, collection: Collection, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Create a new collection.
	Called with `POST /collections`.
	Args:
	collection: the collection
	Returns:
	The collection that was created.
	"""
	...
	@abc.abstractmethod
	def update_collection(
	self, collection_id: str, collection: Collection, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Perform a complete update on an existing collection.
	Called with `PUT /collections/{collection_id}`. It is expected that this
	collection already exists. The update should do a diff against the saved
	collection and perform any necessary updates. Partial updates are not
	supported by the transactions extension.
	Args:
	collection_id: id of the existing collection to be updated
	collection: the updated collection (must be complete)
	Returns:
	The updated collection.
	"""
	...
	@abc.abstractmethod
	def delete_collection(
	self, collection_id: str, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Delete a collection.
	Called with `DELETE /collections/{collection_id}`
	Args:
	collection_id: id of the collection.
	Returns:
	The deleted collection.
	"""
	...
	@attr.s # type:ignore
	class AsyncBaseTransactionsClient(abc.ABC):
	"""Defines a pattern for implementing the STAC transaction extension."""
	@abc.abstractmethod
	async def create_item(
	self,
	collection_id: str,
	item: Union[Item, ItemCollection],
	**kwargs,
	) -> Optional[Union[stac.Item, Response, None]]:
	"""Create a new item.
	Called with `POST /collections/{collection_id}/items`.
	Args:
	item: the item or item collection
	collection_id: the id of the collection from the resource path
	Returns:
	The item that was created or None if item collection.
	"""
	...
	@abc.abstractmethod
	async def update_item(
	self, collection_id: str, item_id: str, item: Item, **kwargs
	) -> Optional[Union[stac.Item, Response]]:
	"""Perform a complete update on an existing item.
	Called with `PUT /collections/{collection_id}/items`. It is expected
	that this item already exists. The update should do a diff against the
	saved item and perform any necessary updates. Partial updates are not
	supported by the transactions extension.
	Args:
	item: the item (must be complete)
	Returns:
	The updated item.
	"""
	...
	@abc.abstractmethod
	async def delete_item(
	self, item_id: str, collection_id: str, **kwargs
	) -> Optional[Union[stac.Item, Response]]:
	"""Delete an item from a collection.
	Called with `DELETE /collections/{collection_id}/items/{item_id}`
	Args:
	item_id: id of the item.
	collection_id: id of the collection.
	Returns:
	The deleted item.
	"""
	...
	@abc.abstractmethod
	async def create_collection(
	self, collection: Collection, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Create a new collection.
	Called with `POST /collections`.
	Args:
	collection: the collection
	Returns:
	The collection that was created.
	"""
	...
	@abc.abstractmethod
	async def update_collection(
	self, collection_id: str, collection: Collection, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Perform a complete update on an existing collection.
	Called with `PUT /collections/{collection_id}`. It is expected that this item
	already exists. The update should do a diff against the saved collection and
	perform any necessary updates. Partial updates are not supported by the
	transactions extension.
	Args:
	collection_id: id of the existing collection to be updated
	collection: the updated collection (must be complete)
	Returns:
	The updated collection.
	"""
	...
	@abc.abstractmethod
	async def delete_collection(
	self, collection_id: str, **kwargs
	) -> Optional[Union[stac.Collection, Response]]:
	"""Delete a collection.
	Called with `DELETE /collections/{collection_id}`
	Args:
	collection_id: id of the collection.
	Returns:
	The deleted collection.
	"""
	...

ref: #744 (comment)_