FileSource: fix requiring all optional file source deps to use a file source (#859)

* fix requiring all optional deps to use any file source

* update documentation for file source

Author: tim-quix

docs/connectors/sources/amazon-s3-source.md

@@ -37,35 +37,50 @@ You can learn more details about the [expected kafka message format](#message-da
 
 ## How To Use
 
-S3 File Source is just a special configuration of the `FileSource` connector.
-
-Simply provide it an `S3Origin` (`FileSource(origin=<ORIGIN>)`).
-
-Then, hand the configured `FileSource` to your `SDF` (`app.dataframe(source=<SOURCE>)`).
+Import and instantiate an `S3FileSource` instance and hand it to an Application using
+`app.add_source(<S3FileSource>)` or instead to a StreamingDataFrame with 
+`app.dataframe(source=<S3FileSource>)` if further data manipulation is required.
 
 For more details around various settings, see [configuration](#configuration).
 
 ```python
 from quixstreams import Application
-from quixstreams.sources.community.file import FileSource
-from quixstreams.sources.community.file.origins import S3Origin
+from quixstreams.sources.community.file.s3 import S3FileSource
+
+
+def key_setter(record: dict) -> str:
+    return record["host_id"]
+
+
+def value_setter(record: dict) -> dict:
+    return {k: record[k] for k in ["field_x", "field_y"]}
 
-app = Application(broker_address="localhost:9092", auto_offset_reset="earliest")
 
-origin = S3Origin(
+def timestamp_setter(record: dict) -> int:
+    return record['timestamp']
+
+
+source = S3FileSource(
+    filepath='folder_a/folder_b',
     bucket="<YOUR BUCKET NAME>",
+    region_name="<YOUR REGION>",
     aws_access_key_id="<YOUR KEY ID>",
     aws_secret_access_key="<YOUR SECRET KEY>",
-    region_name="<YOUR REGION>",
-)
-source = FileSource(
-    directory="path/to/your/topic_folder/",
-    origin=origin,
-    format="json",
+    key_setter=key_setter,
+    value_setter=value_setter,
+    timestamp_setter=timestamp_setter,
+    file_format="json",
     compression="gzip",
+    has_partition_folders=False,
+    replay_speed=0.5,
+)
+app = Application(
+    broker_address="localhost:9092", 
+    consumer_group='file-source',
+    auto_offset_reset='latest',
 )
-sdf = app.dataframe(source=source).print(metadata=True)
-# YOUR LOGIC HERE!
+app.add_source(source)
+
 
 if __name__ == "__main__":
     app.run()
@@ -77,8 +92,9 @@ Here are some important configurations to be aware of (see [File Source API](../
 
 ### Required:
 
-`S3Origin`:
-
+- `filepath`: folder to recursively iterate from (a file will be used directly).  
+    For S3, exclude bucket name or starting "/".    
+    **Note**: If using alongside `FileSink`, provide the path to the topic name folder (ex: `"path/to/topic_a/"`).
 - `bucket`: The S3 bucket name only (ex: `"your-bucket"`).
 - `region_name`: AWS region (ex: us-east-1).    
     **Note**: can alternatively set the `AWS_REGION` environment variable.
@@ -87,18 +103,8 @@ Here are some important configurations to be aware of (see [File Source API](../
 - `aws_secret_access_key`: AWS secret key.    
     **Note**: can alternatively set the `AWS_SECRET_ACCESS_KEY` environment variable.
 
-
-`FileSource`:
-
-- `directory`: a directory to recursively read through (exclude bucket name or starting "/").    
-    **Note**: If using alongside `FileSink`, provide the path to the topic name folder (ex: `"path/to/topic_a/"`).    
-- `origin`: An `S3Origin` instance.
-
-
 ### Optional:
 
-`FileSource`:
-
 - `format`: what format the message files are in (ex: `"json"`, `"parquet"`).    
     **Advanced**: can optionally provide a `Format` instance (`compression` will then be ignored).    
     **Default**: `"json"`
@@ -110,9 +116,21 @@ Here are some important configurations to be aware of (see [File Source API](../
     **Note**: Time delay will only be accurate _per partition_, NOT overall.    
     **Default**: 1.0
 
-## File hierarchy/structure
 
-The File Source expects a folder structure like so:
+## Supported File Hierarchies
+
+All `*FileSource` types support both single file referencing and recursive folder traversal.
+
+In addition, it also supports a topic-partition file structure as produced by a Quix 
+Streams `*FileSink` instance. 
+
+
+### Using with a Topic-Partition hierarchy (from `*FileSink`)
+
+A Topic-Partition structure allows reproducing messages to the exact partition
+they originated from.
+
+When using a Quix Streams `*FileSink`, it will produce files using this structure:
 
 ```
     my_sinked_topics/
@@ -127,6 +145,61 @@ The File Source expects a folder structure like so:
         └── etc...
 ```
 
+To have `*FileSource` reflect this partition mapping for messages (instead of just producing
+messages to whatever partition is applicable), it must know how many partition folders 
+there are so it can create a topic with that many partitions.
+
+To enable this: 
+1. subclass your `*FileSource` instance and define the `file_partition_counter` method.
+    - this will be run before processing any files.
+2. Enable the use of `file_partition_counter` by setting the flag `has_partition_folders=True`.
+3. Extract the original Kafka key with `key_setter` (by default, it uses the same field name that `*FinkSink` writes to).
+   - see [message data schema](#message-data-formatschema) for more info around expected defaults.
+
+#### Example
+As a simple example, using the topic-partition file structure:
+
+```
+├── my_topic/        
+│   ├── 0/          
+│   │   ├── 0000.ext
+│   │   └── 0011.ext
+│   └── 1/
+│       ├── 0003.ext
+│       └── 0016.ext
+```
+
+you could define `file_partition_counter` on `LocalFileSource` like this:
+
+```python
+from quixstreams.sources.community.file.local import LocalFileSource
+
+class MyLocalFileSource(LocalFileSource):
+    
+    def file_partition_counter(self) -> int:
+        return len([f for f in self._filepath.iterdir()])  # `len(['0', '1'])`
+```
+
+Also, for our `key_setter`:
+```python
+def my_key_setter(record: dict) -> str:
+    return record["original_key_field"]
+```
+
+Then when initing with your new class:
+
+```python
+source = MyLocalFileSource(
+    ..., # required args,
+    has_partition_folders=True,
+    key_setter=my_key_setter,
+)
+```
+
+This will produce these messages across the 2 partitions in their original partitioning
+and ordering.
+
+
 ## Message Data Format/Schema
 
 The expected file schema largely depends on the chosen 

docs/connectors/sources/local-file-source.md

@@ -1,3 +1,5 @@
+from quixstreams.sources.community.file.local import LocalFileSource
+
 # Local File Source
 
 !!! info
@@ -23,7 +25,7 @@ pip install quixstreams
 
 ## How It Works
 
-`FileSource` steps through each folder within the provided path and dumps each record 
+`LocalFileSource` steps through each folder within the provided path and dumps each record 
 contained in each file as a message to a Kafka topic. Folders are navigated in 
 lexicographical order.
 
@@ -37,26 +39,46 @@ You can learn more details about the [expected kafka message format](#message-da
 
 ## How To Use
 
-Local File Source is the default configuration of the `FileSource` connector.
-
-Simply hand the configured `FileSource` (without a `origin`) to your `SDF` 
-(`app.dataframe(source=<SOURCE>)`).
+Import and instantiate a `LocalFileSource` instance and hand it to an Application using
+`app.add_source(<LocalFileSource>)` or instead to a StreamingDataFrame with 
+`app.dataframe(source=<LocalFileSource>)` if further data manipulation is required.
 
 For more details around various settings, see [configuration](#configuration).
 
 ```python
 from quixstreams import Application
-from quixstreams.sources.community.file import FileSource
+from quixstreams.sources.community.file.local import LocalFileSource
+
+
+def key_setter(record: dict) -> str:
+    return record["host_id"]
+
+
+def value_setter(record: dict) -> dict:
+    return {k: record[k] for k in ["field_x", "field_y"]}
+
+
+def timestamp_setter(record: dict) -> int:
+    return record['timestamp']
+
 
-app = Application(broker_address="localhost:9092")
-source = FileSource(
-    directory="/path/to/my/topic_folder",
-    format="json",
+source = LocalFileSource(
+    filepath='folder_a/folder_b',
+    key_setter=key_setter,
+    value_setter=value_setter,
+    timestamp_setter=timestamp_setter,
+    file_format="json",
     compression="gzip",
-    replay_speed=1.0,
+    has_partition_folders=False,
+    replay_speed=0.5,
 )
-sdf = app.dataframe(source=source).print(metadata=True)
-# YOUR LOGIC HERE!
+app = Application(
+    broker_address="localhost:9092", 
+    consumer_group='file-source',
+    auto_offset_reset='latest',
+)
+app.add_source(source)
+
 
 if __name__ == "__main__":
     app.run()
@@ -68,7 +90,7 @@ Here are some important configurations to be aware of (see [File Source API](../
 
 ### Required:
 
-- `directory`: a directory to recursively read through (exclude bucket name).    
+- `filepath`: folder to recursively iterate from (a file will be used directly).  
     **Note**: If using alongside `FileSink`, provide the path to the topic name folder (ex: `"path/to/topic_a/"`).
 
 ### Optional:
@@ -85,9 +107,20 @@ Here are some important configurations to be aware of (see [File Source API](../
     **Default**: 1.0
 
 
-## File hierarchy/structure
+## Supported File Hierarchies
+
+All `*FileSource` types support both single file referencing and recursive folder traversal.
+
+In addition, it also supports a topic-partition file structure as produced by a Quix 
+Streams `*FileSink` instance. 
+
 
-The File Source expects a folder structure like so:
+### Using with a Topic-Partition hierarchy (from `*FileSink`)
+
+A Topic-Partition structure allows reproducing messages to the exact partition
+they originated from.
+
+When using a Quix Streams `*FileSink`, it will produce files using this structure:
 
 ```
     my_sinked_topics/
@@ -102,7 +135,60 @@ The File Source expects a folder structure like so:
         └── etc...
 ```
 
-This is the default structure generated by the File Sink.
+To have `*FileSource` reflect this partition mapping for messages (instead of just producing
+messages to whatever partition is applicable), it must know how many partition folders 
+there are so it can create a topic with that many partitions.
+
+To enable this: 
+1. subclass your `*FileSource` instance and define the `file_partition_counter` method.
+    - this will be run before processing any files.
+2. Enable the use of `file_partition_counter` by setting the flag `has_partition_folders=True`.
+3. Extract the original Kafka key with `key_setter` (by default, it uses the same field name that `*FinkSink` writes to).
+   - see [message data schema](#message-data-formatschema) for more info around expected defaults.
+
+#### Example
+As a simple example, using the topic-partition file structure:
+
+```
+├── my_topic/        
+│   ├── 0/          
+│   │   ├── 0000.ext
+│   │   └── 0011.ext
+│   └── 1/
+│       ├── 0003.ext
+│       └── 0016.ext
+```
+
+you could define `file_partition_counter` on `LocalFileSource` like this:
+
+```python
+from quixstreams.sources.community.file.local import LocalFileSource
+
+class MyLocalFileSource(LocalFileSource):
+    
+    def file_partition_counter(self) -> int:
+        return len([f for f in self._filepath.iterdir()])  # `len(['0', '1'])`
+```
+
+Also, for our `key_setter`:
+```python
+def my_key_setter(record: dict) -> str:
+    return record["original_key_field"]
+```
+
+Then when initing with your new class:
+
+```python
+source = MyLocalFileSource(
+    ..., # required args,
+    has_partition_folders=True,
+    key_setter=my_key_setter,
+)
+```
+
+This will produce these messages across the 2 partitions in their original partitioning
+and ordering.
+
 
 ## Message Data Format/Schema