hojichar.core.async_composition 

  1from __future__ import annotations
  2
  3import asyncio
  4import logging
  5from concurrent.futures import Executor, ThreadPoolExecutor
  6from typing import Any, AsyncGenerator, AsyncIterable, Iterable, Sequence
  7
  8import numpy as np
  9
 10from hojichar.core import inspection
 11from hojichar.core.async_filter_interface import AsyncFilter
 12from hojichar.core.composition import Compose
 13from hojichar.core.filter_interface import Filter
 14from hojichar.core.models import Document, Statistics, get_doc_info
 15from hojichar.utils.async_handlers import (
 16    AsyncToSyncIterator,
 17    handle_async_stream_as_sync,
 18    handle_stream_as_async,
 19)
 20
 21
 22class AsyncFilterAdapter(AsyncFilter):
 23    """
 24    Adapter class for executing hojichar.Filter asynchronously.
 25    """
 26
 27    def __init__(
 28        self,
 29        sync_filter: Filter,
 30        *args: Any,
 31        executor: Executor | None = None,
 32        use_batch: bool = True,
 33        **kwargs: Any,
 34    ):
 35        """
 36        Adapter class for executing hojichar.Filter asynchronously.
 37        Used to incorporate Filter into AsyncCompose.
 38
 39        To reduce the overhead of asynchronous context switching,
 40        use_batch is set to True by default to process in batches
 41        in apply_stream, regardless of the sync_filter's use_batch setting.
 42
 43        If performing CPU-bound and heavy processing, you can specify an executor
 44        to offload the processing to the executor. However, due to Python's GIL
 45        constraints, using ThreadPoolExecutor will not parallelize CPU-bound
 46        processing, and the entire process will be locked.
 47
 48        By using ProcessPoolExecutor as the executor, it may be possible to
 49        parallelize CPU-bound processing. However, for parallelizing CPU-bound
 50        processing, it is recommended to use the hojichar.Parallel class to
 51        parallelize synchronous Compose pipeline.
 52        """
 53        super().__init__(*args, use_batch=use_batch, **kwargs)
 54        self.sync_filter = sync_filter
 55        self._has_external_executor = executor is not None
 56        self._executor = executor or ThreadPoolExecutor()
 57        self.batch_size = sync_filter.batch_size
 58
 59    async def apply(self, document: Document) -> Document:
 60        loop = asyncio.get_running_loop()
 61        return await loop.run_in_executor(self._executor, self.sync_filter.apply, document)
 62
 63    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
 64        loop = asyncio.get_running_loop()
 65        return await loop.run_in_executor(
 66            self._executor,
 67            lambda: self.sync_filter.apply_batch(batch),
 68        )
 69
 70    async def shutdown(self) -> None:
 71        self.sync_filter.shutdown()
 72        if not self._has_external_executor:
 73            self._executor.shutdown()
 74
 75
 76class AsyncCompose(AsyncFilter):
 77    def __init__(
 78        self,
 79        filters: list[AsyncFilter | Filter],
 80        random_state: int | np.random.Generator | None = None,
 81        executor: ThreadPoolExecutor | None = None,
 82        *args: Any,
 83        **kwargs: Any,
 84    ):
 85        super().__init__(random_state=random_state, *args, **kwargs)
 86        self.logger = logging.getLogger(f"{self.__module__}.{self.__class__.__name__}")
 87        self._statistics.name = "Total"
 88        self._has_external_executor = executor is not None
 89        self._executor = executor or ThreadPoolExecutor()
 90        self.set_filters(filters)
 91
 92    def set_filters(self, filters: list[AsyncFilter | Filter]) -> None:
 93        self.filters: list[AsyncFilter] = []
 94        filter_idx = 0
 95        for f in filters:
 96            if isinstance(f, (AsyncCompose, Compose)):
 97                for sub in f.filters:
 98                    name = f"{filter_idx}-{sub.__class__.__name__}"
 99                    if isinstance(sub, Filter):
100                        name = f"{filter_idx}-{sub.__class__.__name__}"
101                        sub = AsyncFilterAdapter(sub, executor=self._executor)
102
103                    sub._set_rng_if_not_initialized(self._rng)
104                    sub.name = name
105                    sub._statistics.name = name
106                    self.filters.append(sub)
107                    filter_idx += 1
108            else:
109                name = f"{filter_idx}-{f.__class__.__name__}"
110                if isinstance(f, Filter):
111                    name = f"{filter_idx}-{f.__class__.__name__}"
112                    f = AsyncFilterAdapter(f, executor=self._executor)
113                f._set_rng_if_not_initialized(self._rng)
114                f.name = name
115                f._statistics.name = name
116                self.filters.append(f)
117                filter_idx += 1
118
119    async def apply(self, document: Document) -> Document:
120        stat = get_doc_info(document)
121        for filter_idx, filt in enumerate(self.filters):
122            document = await filt._apply(document)
123        new_stat = get_doc_info(document)
124        async with self._stats_lock:
125            self._statistics.update_by_diff(stat, new_stat)
126        return document
127
128    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
129        stats = [get_doc_info(doc) for doc in batch]
130        for i, filt in enumerate(self.filters):
131            batch = await filt._apply_batch(batch)
132        batch = await self._finalize_batch(batch, stats)
133        return list(batch)
134
135    async def apply_stream(
136        self,
137        stream: AsyncIterable[Document] | Iterable[Document],
138    ) -> AsyncGenerator[Document, None]:
139        async_stream = handle_stream_as_async(stream, chunk_size=1000, executor=self._executor)
140        async_stream = self._count_input_stats(async_stream)
141
142        for i, filt in enumerate(self.filters):
143            async_stream = filt.apply_stream(async_stream)
144
145        async for doc in async_stream:
146            in_stat = doc._get_initial_stats()
147            if in_stat is None:
148                in_stat = get_doc_info(doc)
149                self.logger.debug(
150                    "Initial stats missing for document during async stream aggregation; "
151                    "using current stats as fallback"
152                )
153            out_stat = get_doc_info(doc)
154            async with self._stats_lock:
155                self._statistics.update_by_diff(in_stat, out_stat)
156            doc._clear_initial_stats()
157            yield doc
158
159    def imap_apply(
160        self,
161        stream: AsyncIterable[Document] | Iterable[Document],
162        *,
163        buffer_size: int = 128,
164        shutdown: bool = True,
165    ) -> AsyncToSyncIterator[Document]:
166        """Synchronously consume this asynchronous pipeline.
167
168        The returned iterator runs the pipeline on a dedicated background event loop. It is
169        intended for one-shot use and shuts down the pipeline after consumption by default.
170        Use the iterator as a context manager if iteration may stop early.
171        """
172
173        finalizer = self.shutdown if shutdown else None
174        return handle_async_stream_as_sync(
175            self.apply_stream(stream),
176            buffer_size=buffer_size,
177            finalizer=finalizer,
178        )
179
180    async def _count_input_stats(
181        self, async_stream: AsyncIterable[Document]
182    ) -> AsyncGenerator[Document, None]:
183        async for doc in async_stream:
184            doc._set_initial_stats(get_doc_info(doc))
185            yield doc
186
187    def get_total_statistics(self) -> list[Statistics]:
188        """
189        Get the statistics of the Compose object and sub filters.
190
191        The statistics of the Compose class are stored in an object with the name "Total",
192        and sub-filters's are stored with names in the format {filter_index}-{filter class name}.
193        """
194        stats = []
195        stats.append(self.get_statistics())
196        for i, filt in enumerate(self.filters):
197            stats.append(filt.get_statistics())
198        return stats
199
200    def get_total_statistics_map(self) -> list[dict[str, Any]]:
201        """
202        Get the statistics of the Compose object and sub filters as a list of dictionaries.
203        """
204        stats = self.get_total_statistics()
205        return [stat.to_dict() for stat in stats]
206
207    @property
208    def statistics(self) -> dict:
209        """
210        Deprecated
211
212        Get the statistics of the Compose object and sub filters.
213
214        This property is retained for compatibility with previous versions.
215        Please use `get_total_statistics` or `get_total_statistics_map` instead.
216        """
217        return inspection.statistics_obj_adapter(  # type: ignore
218            self.get_total_statistics()
219        ).get_human_readable_values()
220
221    @property
222    def statistics_obj(self) -> inspection.StatsContainer:
223        """
224        Deprecated
225
226        Get the statistics of the AsyncCompose object and sub filters.
227        This method returns a StatsContainer object which contains the statistics
228        of the AsyncCompose object and sub filters.
229
230        This property is retained for compatibility with previous versions.
231        Please use `get_total_statistics` or `get_total_statistics_map` instead.
232        """
233        return inspection.statistics_obj_adapter(self.get_total_statistics())  # type: ignore
234
235    async def shutdown(self) -> None:
236        for filt in self.filters:
237            await filt.shutdown()
238        if not self._has_external_executor:
239            self._executor.shutdown()
240
241    async def __aenter__(self) -> "AsyncCompose":
242        return self
243
244    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
245        await self.shutdown()
246        if exc_type is not None:
247            raise exc_value
class AsyncFilterAdapter(hojichar.core.async_filter_interface.AsyncFilter): 
23class AsyncFilterAdapter(AsyncFilter):
24    """
25    Adapter class for executing hojichar.Filter asynchronously.
26    """
27
28    def __init__(
29        self,
30        sync_filter: Filter,
31        *args: Any,
32        executor: Executor | None = None,
33        use_batch: bool = True,
34        **kwargs: Any,
35    ):
36        """
37        Adapter class for executing hojichar.Filter asynchronously.
38        Used to incorporate Filter into AsyncCompose.
39
40        To reduce the overhead of asynchronous context switching,
41        use_batch is set to True by default to process in batches
42        in apply_stream, regardless of the sync_filter's use_batch setting.
43
44        If performing CPU-bound and heavy processing, you can specify an executor
45        to offload the processing to the executor. However, due to Python's GIL
46        constraints, using ThreadPoolExecutor will not parallelize CPU-bound
47        processing, and the entire process will be locked.
48
49        By using ProcessPoolExecutor as the executor, it may be possible to
50        parallelize CPU-bound processing. However, for parallelizing CPU-bound
51        processing, it is recommended to use the hojichar.Parallel class to
52        parallelize synchronous Compose pipeline.
53        """
54        super().__init__(*args, use_batch=use_batch, **kwargs)
55        self.sync_filter = sync_filter
56        self._has_external_executor = executor is not None
57        self._executor = executor or ThreadPoolExecutor()
58        self.batch_size = sync_filter.batch_size
59
60    async def apply(self, document: Document) -> Document:
61        loop = asyncio.get_running_loop()
62        return await loop.run_in_executor(self._executor, self.sync_filter.apply, document)
63
64    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
65        loop = asyncio.get_running_loop()
66        return await loop.run_in_executor(
67            self._executor,
68            lambda: self.sync_filter.apply_batch(batch),
69        )
70
71    async def shutdown(self) -> None:
72        self.sync_filter.shutdown()
73        if not self._has_external_executor:
74            self._executor.shutdown()

Adapter class for executing hojichar.Filter asynchronously.

AsyncFilterAdapter( sync_filter: hojichar.core.filter_interface.Filter, *args: Any, executor: concurrent.futures._base.Executor | None = None, use_batch: bool = True, **kwargs: Any) 
28    def __init__(
29        self,
30        sync_filter: Filter,
31        *args: Any,
32        executor: Executor | None = None,
33        use_batch: bool = True,
34        **kwargs: Any,
35    ):
36        """
37        Adapter class for executing hojichar.Filter asynchronously.
38        Used to incorporate Filter into AsyncCompose.
39
40        To reduce the overhead of asynchronous context switching,
41        use_batch is set to True by default to process in batches
42        in apply_stream, regardless of the sync_filter's use_batch setting.
43
44        If performing CPU-bound and heavy processing, you can specify an executor
45        to offload the processing to the executor. However, due to Python's GIL
46        constraints, using ThreadPoolExecutor will not parallelize CPU-bound
47        processing, and the entire process will be locked.
48
49        By using ProcessPoolExecutor as the executor, it may be possible to
50        parallelize CPU-bound processing. However, for parallelizing CPU-bound
51        processing, it is recommended to use the hojichar.Parallel class to
52        parallelize synchronous Compose pipeline.
53        """
54        super().__init__(*args, use_batch=use_batch, **kwargs)
55        self.sync_filter = sync_filter
56        self._has_external_executor = executor is not None
57        self._executor = executor or ThreadPoolExecutor()
58        self.batch_size = sync_filter.batch_size

Adapter class for executing hojichar.Filter asynchronously. Used to incorporate Filter into AsyncCompose.

To reduce the overhead of asynchronous context switching, use_batch is set to True by default to process in batches in apply_stream, regardless of the sync_filter's use_batch setting.

If performing CPU-bound and heavy processing, you can specify an executor to offload the processing to the executor. However, due to Python's GIL constraints, using ThreadPoolExecutor will not parallelize CPU-bound processing, and the entire process will be locked.

By using ProcessPoolExecutor as the executor, it may be possible to parallelize CPU-bound processing. However, for parallelizing CPU-bound processing, it is recommended to use the hojichar.Parallel class to parallelize synchronous Compose pipeline.

async def apply( self, document: hojichar.core.models.Document) -> hojichar.core.models.Document: 
60    async def apply(self, document: Document) -> Document:
61        loop = asyncio.get_running_loop()
62        return await loop.run_in_executor(self._executor, self.sync_filter.apply, document)

Definition of async filter behavior.

In this method, the filter will modify document.text or document.extras and set document.is_rejected = True to discard the document.

Parameters

document : Document Input document

Returns

Document Processed Document

async def apply_batch( self, batch: Sequence[hojichar.core.models.Document]) -> list[hojichar.core.models.Document]: 
64    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
65        loop = asyncio.get_running_loop()
66        return await loop.run_in_executor(
67            self._executor,
68            lambda: self.sync_filter.apply_batch(batch),
69        )

Apply the filter to a Sequence of documents. By default, the processing implemented in apply is executed asynchronously and concurrently. If the filter processing can be optimized for batch processing, override this method.

async def shutdown(self) -> None: 
71    async def shutdown(self) -> None:
72        self.sync_filter.shutdown()
73        if not self._has_external_executor:
74            self._executor.shutdown()

You can override this method to release resources or perform cleanup tasks.

Inherited Members
hojichar.core.async_filter_interface.AsyncFilter
apply_stream
get_statistics
get_statistics_map
get_jsonable_vars
class AsyncCompose(hojichar.core.async_filter_interface.AsyncFilter): 
 77class AsyncCompose(AsyncFilter):
 78    def __init__(
 79        self,
 80        filters: list[AsyncFilter | Filter],
 81        random_state: int | np.random.Generator | None = None,
 82        executor: ThreadPoolExecutor | None = None,
 83        *args: Any,
 84        **kwargs: Any,
 85    ):
 86        super().__init__(random_state=random_state, *args, **kwargs)
 87        self.logger = logging.getLogger(f"{self.__module__}.{self.__class__.__name__}")
 88        self._statistics.name = "Total"
 89        self._has_external_executor = executor is not None
 90        self._executor = executor or ThreadPoolExecutor()
 91        self.set_filters(filters)
 92
 93    def set_filters(self, filters: list[AsyncFilter | Filter]) -> None:
 94        self.filters: list[AsyncFilter] = []
 95        filter_idx = 0
 96        for f in filters:
 97            if isinstance(f, (AsyncCompose, Compose)):
 98                for sub in f.filters:
 99                    name = f"{filter_idx}-{sub.__class__.__name__}"
100                    if isinstance(sub, Filter):
101                        name = f"{filter_idx}-{sub.__class__.__name__}"
102                        sub = AsyncFilterAdapter(sub, executor=self._executor)
103
104                    sub._set_rng_if_not_initialized(self._rng)
105                    sub.name = name
106                    sub._statistics.name = name
107                    self.filters.append(sub)
108                    filter_idx += 1
109            else:
110                name = f"{filter_idx}-{f.__class__.__name__}"
111                if isinstance(f, Filter):
112                    name = f"{filter_idx}-{f.__class__.__name__}"
113                    f = AsyncFilterAdapter(f, executor=self._executor)
114                f._set_rng_if_not_initialized(self._rng)
115                f.name = name
116                f._statistics.name = name
117                self.filters.append(f)
118                filter_idx += 1
119
120    async def apply(self, document: Document) -> Document:
121        stat = get_doc_info(document)
122        for filter_idx, filt in enumerate(self.filters):
123            document = await filt._apply(document)
124        new_stat = get_doc_info(document)
125        async with self._stats_lock:
126            self._statistics.update_by_diff(stat, new_stat)
127        return document
128
129    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
130        stats = [get_doc_info(doc) for doc in batch]
131        for i, filt in enumerate(self.filters):
132            batch = await filt._apply_batch(batch)
133        batch = await self._finalize_batch(batch, stats)
134        return list(batch)
135
136    async def apply_stream(
137        self,
138        stream: AsyncIterable[Document] | Iterable[Document],
139    ) -> AsyncGenerator[Document, None]:
140        async_stream = handle_stream_as_async(stream, chunk_size=1000, executor=self._executor)
141        async_stream = self._count_input_stats(async_stream)
142
143        for i, filt in enumerate(self.filters):
144            async_stream = filt.apply_stream(async_stream)
145
146        async for doc in async_stream:
147            in_stat = doc._get_initial_stats()
148            if in_stat is None:
149                in_stat = get_doc_info(doc)
150                self.logger.debug(
151                    "Initial stats missing for document during async stream aggregation; "
152                    "using current stats as fallback"
153                )
154            out_stat = get_doc_info(doc)
155            async with self._stats_lock:
156                self._statistics.update_by_diff(in_stat, out_stat)
157            doc._clear_initial_stats()
158            yield doc
159
160    def imap_apply(
161        self,
162        stream: AsyncIterable[Document] | Iterable[Document],
163        *,
164        buffer_size: int = 128,
165        shutdown: bool = True,
166    ) -> AsyncToSyncIterator[Document]:
167        """Synchronously consume this asynchronous pipeline.
168
169        The returned iterator runs the pipeline on a dedicated background event loop. It is
170        intended for one-shot use and shuts down the pipeline after consumption by default.
171        Use the iterator as a context manager if iteration may stop early.
172        """
173
174        finalizer = self.shutdown if shutdown else None
175        return handle_async_stream_as_sync(
176            self.apply_stream(stream),
177            buffer_size=buffer_size,
178            finalizer=finalizer,
179        )
180
181    async def _count_input_stats(
182        self, async_stream: AsyncIterable[Document]
183    ) -> AsyncGenerator[Document, None]:
184        async for doc in async_stream:
185            doc._set_initial_stats(get_doc_info(doc))
186            yield doc
187
188    def get_total_statistics(self) -> list[Statistics]:
189        """
190        Get the statistics of the Compose object and sub filters.
191
192        The statistics of the Compose class are stored in an object with the name "Total",
193        and sub-filters's are stored with names in the format {filter_index}-{filter class name}.
194        """
195        stats = []
196        stats.append(self.get_statistics())
197        for i, filt in enumerate(self.filters):
198            stats.append(filt.get_statistics())
199        return stats
200
201    def get_total_statistics_map(self) -> list[dict[str, Any]]:
202        """
203        Get the statistics of the Compose object and sub filters as a list of dictionaries.
204        """
205        stats = self.get_total_statistics()
206        return [stat.to_dict() for stat in stats]
207
208    @property
209    def statistics(self) -> dict:
210        """
211        Deprecated
212
213        Get the statistics of the Compose object and sub filters.
214
215        This property is retained for compatibility with previous versions.
216        Please use `get_total_statistics` or `get_total_statistics_map` instead.
217        """
218        return inspection.statistics_obj_adapter(  # type: ignore
219            self.get_total_statistics()
220        ).get_human_readable_values()
221
222    @property
223    def statistics_obj(self) -> inspection.StatsContainer:
224        """
225        Deprecated
226
227        Get the statistics of the AsyncCompose object and sub filters.
228        This method returns a StatsContainer object which contains the statistics
229        of the AsyncCompose object and sub filters.
230
231        This property is retained for compatibility with previous versions.
232        Please use `get_total_statistics` or `get_total_statistics_map` instead.
233        """
234        return inspection.statistics_obj_adapter(self.get_total_statistics())  # type: ignore
235
236    async def shutdown(self) -> None:
237        for filt in self.filters:
238            await filt.shutdown()
239        if not self._has_external_executor:
240            self._executor.shutdown()
241
242    async def __aenter__(self) -> "AsyncCompose":
243        return self
244
245    async def __aexit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
246        await self.shutdown()
247        if exc_type is not None:
248            raise exc_value

Helper class that provides a standard way to create an ABC using inheritance.

AsyncCompose( filters: list[hojichar.core.async_filter_interface.AsyncFilter | hojichar.core.filter_interface.Filter], random_state: int | numpy.random._generator.Generator | None = None, executor: concurrent.futures.thread.ThreadPoolExecutor | None = None, *args: Any, **kwargs: Any) 
78    def __init__(
79        self,
80        filters: list[AsyncFilter | Filter],
81        random_state: int | np.random.Generator | None = None,
82        executor: ThreadPoolExecutor | None = None,
83        *args: Any,
84        **kwargs: Any,
85    ):
86        super().__init__(random_state=random_state, *args, **kwargs)
87        self.logger = logging.getLogger(f"{self.__module__}.{self.__class__.__name__}")
88        self._statistics.name = "Total"
89        self._has_external_executor = executor is not None
90        self._executor = executor or ThreadPoolExecutor()
91        self.set_filters(filters)

Base class for asynchronous filters.

Parameters

p : float The probability of applying the filter. If p is 1, the filter will always be applied. skip_rejected : bool If True, the filter will skip documents that are already rejected. If you want to apply the filter to all documents (e.g., postprocess), set this to False. random_state : Optional[Union[int, np.random.Generator]] Seed for the random number generator. If None is specified, the random number generator managed by the Compose class will be used. use_batch : bool If True, the filter will process documents in batches in the apply_stream method. batch_size : int The size of the batch to process documents in the apply_stream method. When apply_batch is not overridden, this is also the maximum number of in-flight document tasks used by the sliding-window scheduler. Ordered processing applies backpressure after at most two windows of started but not yet yielded documents. ordered : bool If True, apply_stream yields documents in input order. If False, filters using the default apply_batch implementation yield documents as soon as their processing completes.

def set_filters( self, filters: list[hojichar.core.async_filter_interface.AsyncFilter | hojichar.core.filter_interface.Filter]) -> None: 
 93    def set_filters(self, filters: list[AsyncFilter | Filter]) -> None:
 94        self.filters: list[AsyncFilter] = []
 95        filter_idx = 0
 96        for f in filters:
 97            if isinstance(f, (AsyncCompose, Compose)):
 98                for sub in f.filters:
 99                    name = f"{filter_idx}-{sub.__class__.__name__}"
100                    if isinstance(sub, Filter):
101                        name = f"{filter_idx}-{sub.__class__.__name__}"
102                        sub = AsyncFilterAdapter(sub, executor=self._executor)
103
104                    sub._set_rng_if_not_initialized(self._rng)
105                    sub.name = name
106                    sub._statistics.name = name
107                    self.filters.append(sub)
108                    filter_idx += 1
109            else:
110                name = f"{filter_idx}-{f.__class__.__name__}"
111                if isinstance(f, Filter):
112                    name = f"{filter_idx}-{f.__class__.__name__}"
113                    f = AsyncFilterAdapter(f, executor=self._executor)
114                f._set_rng_if_not_initialized(self._rng)
115                f.name = name
116                f._statistics.name = name
117                self.filters.append(f)
118                filter_idx += 1
async def apply( self, document: hojichar.core.models.Document) -> hojichar.core.models.Document: 
120    async def apply(self, document: Document) -> Document:
121        stat = get_doc_info(document)
122        for filter_idx, filt in enumerate(self.filters):
123            document = await filt._apply(document)
124        new_stat = get_doc_info(document)
125        async with self._stats_lock:
126            self._statistics.update_by_diff(stat, new_stat)
127        return document

Definition of async filter behavior.

In this method, the filter will modify document.text or document.extras and set document.is_rejected = True to discard the document.

Parameters

document : Document Input document

Returns

Document Processed Document

async def apply_batch( self, batch: Sequence[hojichar.core.models.Document]) -> list[hojichar.core.models.Document]: 
129    async def apply_batch(self, batch: Sequence[Document]) -> list[Document]:
130        stats = [get_doc_info(doc) for doc in batch]
131        for i, filt in enumerate(self.filters):
132            batch = await filt._apply_batch(batch)
133        batch = await self._finalize_batch(batch, stats)
134        return list(batch)

Apply the filter to a Sequence of documents. By default, the processing implemented in apply is executed asynchronously and concurrently. If the filter processing can be optimized for batch processing, override this method.

async def apply_stream( self, stream: Union[AsyncIterable[hojichar.core.models.Document], Iterable[hojichar.core.models.Document]]) -> AsyncGenerator[hojichar.core.models.Document, NoneType]: 
136    async def apply_stream(
137        self,
138        stream: AsyncIterable[Document] | Iterable[Document],
139    ) -> AsyncGenerator[Document, None]:
140        async_stream = handle_stream_as_async(stream, chunk_size=1000, executor=self._executor)
141        async_stream = self._count_input_stats(async_stream)
142
143        for i, filt in enumerate(self.filters):
144            async_stream = filt.apply_stream(async_stream)
145
146        async for doc in async_stream:
147            in_stat = doc._get_initial_stats()
148            if in_stat is None:
149                in_stat = get_doc_info(doc)
150                self.logger.debug(
151                    "Initial stats missing for document during async stream aggregation; "
152                    "using current stats as fallback"
153                )
154            out_stat = get_doc_info(doc)
155            async with self._stats_lock:
156                self._statistics.update_by_diff(in_stat, out_stat)
157            doc._clear_initial_stats()
158            yield doc

Apply the filter to a stream of documents (Iterable or AsyncIterable). If use_batch is set to True at initialization, the filter will process documents in batches. If the stream is not asynchronous, use handle_stream_as_async to convert it to an asynchronous stream.

Even if an exception occurs during processing, the process will continue, and the following actions will be taken:

  • Set the is_rejected flag of the document to True
  • Set the error details in reject_reason
  • Increment the errors count in the statistics retrievable via get_statistics
def imap_apply( self, stream: Union[AsyncIterable[hojichar.core.models.Document], Iterable[hojichar.core.models.Document]], *, buffer_size: int = 128, shutdown: bool = True) -> hojichar.utils.async_handlers.AsyncToSyncIterator[hojichar.core.models.Document]: 
160    def imap_apply(
161        self,
162        stream: AsyncIterable[Document] | Iterable[Document],
163        *,
164        buffer_size: int = 128,
165        shutdown: bool = True,
166    ) -> AsyncToSyncIterator[Document]:
167        """Synchronously consume this asynchronous pipeline.
168
169        The returned iterator runs the pipeline on a dedicated background event loop. It is
170        intended for one-shot use and shuts down the pipeline after consumption by default.
171        Use the iterator as a context manager if iteration may stop early.
172        """
173
174        finalizer = self.shutdown if shutdown else None
175        return handle_async_stream_as_sync(
176            self.apply_stream(stream),
177            buffer_size=buffer_size,
178            finalizer=finalizer,
179        )

Synchronously consume this asynchronous pipeline.

The returned iterator runs the pipeline on a dedicated background event loop. It is intended for one-shot use and shuts down the pipeline after consumption by default. Use the iterator as a context manager if iteration may stop early.

def get_total_statistics(self) -> list[hojichar.core.models.Statistics]: 
188    def get_total_statistics(self) -> list[Statistics]:
189        """
190        Get the statistics of the Compose object and sub filters.
191
192        The statistics of the Compose class are stored in an object with the name "Total",
193        and sub-filters's are stored with names in the format {filter_index}-{filter class name}.
194        """
195        stats = []
196        stats.append(self.get_statistics())
197        for i, filt in enumerate(self.filters):
198            stats.append(filt.get_statistics())
199        return stats

Get the statistics of the Compose object and sub filters.

The statistics of the Compose class are stored in an object with the name "Total", and sub-filters's are stored with names in the format {filter_index}-{filter class name}.

def get_total_statistics_map(self) -> list[dict[str, typing.Any]]: 
201    def get_total_statistics_map(self) -> list[dict[str, Any]]:
202        """
203        Get the statistics of the Compose object and sub filters as a list of dictionaries.
204        """
205        stats = self.get_total_statistics()
206        return [stat.to_dict() for stat in stats]

Get the statistics of the Compose object and sub filters as a list of dictionaries.

statistics: dict

Deprecated

Get the statistics of the Compose object and sub filters.

This property is retained for compatibility with previous versions. Please use get_total_statistics or get_total_statistics_map instead.

statistics_obj: hojichar.core.inspection.StatsContainer

Deprecated

Get the statistics of the AsyncCompose object and sub filters. This method returns a StatsContainer object which contains the statistics of the AsyncCompose object and sub filters.

This property is retained for compatibility with previous versions. Please use get_total_statistics or get_total_statistics_map instead.

async def shutdown(self) -> None: 
236    async def shutdown(self) -> None:
237        for filt in self.filters:
238            await filt.shutdown()
239        if not self._has_external_executor:
240            self._executor.shutdown()

You can override this method to release resources or perform cleanup tasks.

Inherited Members
hojichar.core.async_filter_interface.AsyncFilter
get_statistics
get_statistics_map
get_jsonable_vars