FlowBuilder API
===============

The FlowBuilder provides a fluent interface for constructing data flows.

Creating a Flow
---------------

Use the :func:`flow.flow` function to create a new FlowBuilder:

.. code-block:: python

   from flow import flow
   
   # Create a new flow
   builder = flow("My Pipeline")
   
   # Create with namespace
   builder = flow("pipeline:data_processor")

FlowBuilder Methods
-------------------

Source Methods
~~~~~~~~~~~~~~

.. method:: FlowBuilder.source(data, data_type, name=None)

   Add a source node to the flow.
   
   :param data: The data source (iterable, callable, or async iterator)
   :param data_type: The type of data produced
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # From list
      builder.source([1, 2, 3], int)
      
      # From generator
      def generate():
          for i in range(10):
              yield i
      
      builder.source(generate(), int)
      
      # From async source
      async def fetch_data():
          async for item in api_stream():
              yield item
      
      builder.source(fetch_data(), dict)

Transform Methods
~~~~~~~~~~~~~~~~~

.. method:: FlowBuilder.transform(func, output_type, name=None)

   Add a transformation to the flow.
   
   :param func: Function to transform each item
   :param output_type: The type of data produced
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # Simple transform
      builder.transform(lambda x: x * 2, int)
      
      # Named function
      def process_data(item):
          return {"value": item, "processed": True}
      
      builder.transform(process_data, dict)

Filter Methods
~~~~~~~~~~~~~~

.. method:: FlowBuilder.filter(predicate, name=None)

   Filter items in the flow.
   
   :param predicate: Function returning True to keep items
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # Filter even numbers
      builder.filter(lambda x: x % 2 == 0)
      
      # Complex filter
      def is_valid(item):
          return item["status"] == "active" and item["value"] > 0
      
      builder.filter(is_valid)

Side Effects
~~~~~~~~~~~~

.. method:: FlowBuilder.tap(func, name=None)

   Add a side effect without transforming the data.
   
   :param func: Function to execute for side effects
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # Logging
      builder.tap(lambda x: print(f"Processing: {x}"))
      
      # Metrics
      metrics = []
      builder.tap(metrics.append)

Sink Methods
~~~~~~~~~~~~

.. method:: FlowBuilder.sink(func, name=None)
.. method:: FlowBuilder.to(func, name=None)

   Add a sink node to consume data. ``to`` is an alias for ``sink``.
   
   :param func: Function to consume each item
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # Collect results
      results = []
      builder.to(results.append)
      
      # Write to file
      async def write_to_file(item):
          async with aiofiles.open("output.txt", "a") as f:
              await f.write(f"{item}\n")
      
      builder.to(write_to_file)

Flow Control
~~~~~~~~~~~~

.. method:: FlowBuilder.split(count=2, name=None)

   Split the flow into multiple outputs.
   
   :param count: Number of output flows
   :param name: Optional name for the node
   :return: List of FlowBuilders
   
   Example:
   
   .. code-block:: python
   
      # Split into two streams
      flow1, flow2 = builder.split(2)
      
      # Process each stream differently
      flow1.transform(lambda x: x * 2, int).to(results1.append)
      flow2.filter(lambda x: x > 5).to(results2.append)

.. method:: FlowBuilder.merge_with(*others, name=None)

   Merge multiple flows together.
   
   :param others: Other FlowBuilders to merge
   :param name: Optional name for the node
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      # Create separate flows
      flow1 = flow("source1").source(data1, int)
      flow2 = flow("source2").source(data2, int)
      
      # Merge them
      flow1.merge_with(flow2).to(results.append)

Middleware
~~~~~~~~~~

.. method:: FlowBuilder.with_middleware(*middleware)

   Add middleware to all nodes in the flow.
   
   :param middleware: Middleware instances to add
   :return: FlowBuilder for chaining
   
   Example:
   
   .. code-block:: python
   
      from flow import LoggingMiddleware, RetryMiddleware
      
      logger = LoggingMiddleware()
      retry = RetryMiddleware(max_attempts=3)
      
      builder.with_middleware(logger, retry)

Execution
~~~~~~~~~

.. method:: FlowBuilder.build()

   Build the flow into an ExecutableGraph.
   
   :return: ExecutableGraph instance
   
   Example:
   
   .. code-block:: python
   
      graph = builder.build()
      await graph.run()

.. method:: FlowBuilder.execute(duration=None, auto_stop=None)

   Build and execute the flow.
   
   :param duration: Maximum duration in seconds
   :param auto_stop: Whether to stop when sources complete
   :return: ExecutableGraph instance
   
   Example:
   
   .. code-block:: python
   
      # Run for 10 seconds
      await builder.execute(duration=10.0)
      
      # Run until completion
      await builder.execute(auto_stop=True)
      
      # Run indefinitely (server mode)
      graph = await builder.execute(auto_stop=False)
      # ... later ...
      await graph.stop()

Complete Example
----------------

.. code-block:: python

   from flow import flow, LoggingMiddleware
   import asyncio
   
   async def main():
       results = []
       
       # Build a complete pipeline
       await (
           flow("ETL Pipeline")
           .with_middleware(LoggingMiddleware())
           .source(range(1, 11), int)
           .filter(lambda x: x % 2 == 0)
           .transform(lambda x: x ** 2, int)
           .tap(lambda x: print(f"Squared: {x}"))
           .to(results.append)
           .execute()
       )
       
       print(f"Results: {results}")
   
   asyncio.run(main())