Custom Generators

from collections.abc import Generator

class IntAndCharSequence(Generator):
    
    def __init__(self) -> None:
        
        self._item = {
            "integer": 0,
            "character": "a"
        }
        
    def send(self, value):
        
        item = self._item.copy()
        
        if item["integer"] == 10:
            raise StopIteration
        
        self._item["integer"] += 1
        self._item["character"] = chr(ord(self._item["character"]) + 1)
        
        # import time
        # time.sleep(0.1)
        
        return item
        
    def throw(self, typ, val=None, tb=None):
        """Raise an exception in the generator.
        Return next yielded value or raise StopIteration.
        
        Notes
        -----
            This implementation is copied from https://github.com/python/cpython/blob/d5d3249e8a37936d32266fa06ac20017307a1f70/Lib/_collections_abc.py#L309.
        """
        
        if val is None:
            if tb is None:
                raise typ
            val = typ()
            
        if tb is not None:
            val = val.with_traceback(tb)
            
        raise val
    

for i, item in enumerate(IntAndCharSequence()):
    print(item)
    if i == 20:
        break

{'integer': 0, 'character': 'a'}
{'integer': 1, 'character': 'b'}
{'integer': 2, 'character': 'c'}
{'integer': 3, 'character': 'd'}
{'integer': 4, 'character': 'e'}
{'integer': 5, 'character': 'f'}
{'integer': 6, 'character': 'g'}
{'integer': 7, 'character': 'h'}
{'integer': 8, 'character': 'i'}
{'integer': 9, 'character': 'j'}

import operator
import functools

operator.itemgetter("integer")

operator.itemgetter('integer')

for i, item in enumerate(map(lambda item: item.get("integer"), IntAndCharSequence())):
    print(item)
    if i == 9:
        break

0
1
2
3
4
5
6
7
8
9

for i, item in enumerate(map(operator.itemgetter("integer"), IntAndCharSequence())):
    print(item)
    if i == 9:
        break

0
1
2
3
4
5
6
7
8
9

g = map(lambda item: item.get("integer"), IntAndCharSequence())

isinstance(g, Generator)

False

from operator import getitem
from functools import partial

book = {
    "title": "A Python Notebook",
    "chapters": [
        {
            "title": "Foundamentals",
            "sections": [
                "Custom Generators"
            ]
        }
    ]
}

getitem(getitem(book, "chapters"), 0)

{'title': 'Foundamentals', 'sections': ['Custom Generators']}

partial(getitem, "chapters")

functools.partial(<built-in function getitem>, 'chapters')

from typing import Self
from abc import ABC, abstractmethod
from typing import Any

class PipelineComponent(ABC):
    
    def __call__(self, input: Any) -> Any:
        
        output = self.forward(input)
        
        return output
        
    def __or__(self, other: Self) -> Self:
        
        this = self
        
        class _CombinedComponent(PipelineComponent):
            
            def forward(self, input: Any) -> Any:
                return other.forward(this.forward(input))
        
        component = _CombinedComponent()
        
        return component
    
    @abstractmethod
    def forward(self, input: Any) -> Any:
        pass

class GetItem(PipelineComponent):
    
    def __init__(self, key: Any) -> None:
        
        super().__init__()
        
        self._key = key
    
    def forward(self, input: Any) -> Any:
        
        return getitem(input, self._key)
    

(GetItem("chapters") | GetItem(0) | GetItem("sections") | GetItem(0))(book)

'Custom Generators'

(GetItem("chapters") | GetItem(0) | GetItem("sections"))

<__main__.PipelineComponent.__or__.<locals>._CombinedComponent at 0x104674a90>

Asynchronous Generators

https://gist.github.com/jspahrsummers/32a8096667cf9f17d5e8fddeb081b202