Add options to store rendered sources and call sequences

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "breakshaft"
-version = "0.1.1.post1"
+version = "0.1.2"
 description = "Library for in-time codegen for type conversion"
 authors = [
     { name = "nikto_b", email = "niktob560@yandex.ru" }

src/breakshaft/convertor.py

@@ -15,10 +15,14 @@ class ConvRepo:
 
     walker: GraphWalker
     renderer: ConvertorRenderer
+    store_callseq: bool
+    store_sources: bool
 
     def __init__(self,
                  graph_walker: Optional[GraphWalker] = None,
-                 renderer: Optional[ConvertorRenderer] = None, ):
+                 renderer: Optional[ConvertorRenderer] = None,
+                 store_callseq: bool = False,
+                 store_sources: bool = False):
         if graph_walker is None:
             graph_walker = GraphWalker()
         if renderer is None:
@@ -27,6 +31,8 @@ class ConvRepo:
         self._convertor_set = set()
         self.walker = graph_walker
         self.renderer = renderer
+        self.store_callseq = store_callseq
+        self.store_sources = store_sources
 
     def create_pipeline(self,
                         from_types: Sequence[type],
@@ -51,7 +57,13 @@ class ConvRepo:
             if injects is not None:
                 from_types += (injects,)
 
-        return self.renderer.render(orig_from_types, pipeline_callseq, force_async=force_async)
+        ret_fn = self.renderer.render(orig_from_types,
+                                      pipeline_callseq,
+                                      force_async=force_async,
+                                      store_sources=self.store_sources)
+        if self.store_callseq:
+            setattr(ret_fn, '__breakshaft_callseq__', pipeline_callseq)
+        return ret_fn
 
     @property
     def convertor_set(self):
@@ -84,7 +96,8 @@ class ConvRepo:
 
     def get_callseq(self,
                     injectors: frozenset[ConversionPoint],
-                    from_types: frozenset[type], fn: Callable,
+                    from_types: frozenset[type],
+                    fn: Callable,
                     force_commutative: bool) -> list[ConversionPoint]:
 
         cg = self.walker.generate_callgraph(injectors, from_types, fn)
@@ -120,7 +133,10 @@ class ConvRepo:
         filtered_injectors = self.filtered_injectors(allow_async, allow_sync)
         callseq = self.get_callseq(filtered_injectors, frozenset(from_types), fn, force_commutative)
 
-        return self.renderer.render(from_types, callseq, force_async=force_async)
+        ret_fn = self.renderer.render(from_types, callseq, force_async=force_async, store_sources=self.store_sources)
+        if self.store_callseq:
+            setattr(ret_fn, '__breakshaft_callseq__', callseq)
+        return ret_fn
 
     def mark_injector(self, *, rettype: Optional[type] = None):
         def inner(func: Callable):

src/breakshaft/renderer.py

@@ -14,7 +14,8 @@ class ConvertorRenderer(Protocol):
     def render(self,
                from_types: Sequence[type],
                callseq: Sequence[ConversionPoint],
-               force_async: bool = False) -> Callable:
+               force_async: bool = False,
+               store_sources: bool = False) -> Callable:
         raise NotImplementedError()
 
 
@@ -91,7 +92,8 @@ class InTimeGenerationConvertorRenderer(ConvertorRenderer):
     def render(self,
                from_types: Sequence[type],
                callseq: Sequence[ConversionPoint],
-               force_async: bool = False) -> Callable:
+               force_async: bool = False,
+               store_sources: bool = False) -> Callable:
 
         fnmap = {}
         conversion_models: list[ConversionRenderData] = []
@@ -128,7 +130,10 @@ class InTimeGenerationConvertorRenderer(ConvertorRenderer):
             is_async=is_async,
         )
         convertor_functext = '\n'.join(list(filter(lambda x: len(x.strip()), convertor_functext.split('\n'))))
+        convertor_functext = convertor_functext.replace(', )', ')').replace(',)', ')')
         exec(convertor_functext, namespace)
         unwrap_func = namespace['convertor']
+        if store_sources:
+            setattr(unwrap_func, '__breakshaft_render_src__', convertor_functext)
 
         return typing.cast(Callable, unwrap_func)