(
function_name: str,
params: List[str],
stmts: List[str],
context: dict,
)
| 118 | |
| 119 | |
| 120 | def compile_function( |
| 121 | function_name: str, |
| 122 | params: List[str], |
| 123 | stmts: List[str], |
| 124 | context: dict, |
| 125 | ): |
| 126 | from pyfory import ENABLE_FORY_CYTHON_SERIALIZATION |
| 127 | |
| 128 | if ENABLE_FORY_CYTHON_SERIALIZATION: |
| 129 | from pyfory import serialization |
| 130 | |
| 131 | context["write_nullable_pybool"] = serialization.write_nullable_pybool |
| 132 | context["read_nullable_pybool"] = serialization.read_nullable_pybool |
| 133 | context["write_nullable_int8"] = serialization.write_nullable_int8 |
| 134 | context["read_nullable_int8"] = serialization.read_nullable_int8 |
| 135 | context["write_nullable_int16"] = serialization.write_nullable_int16 |
| 136 | context["read_nullable_int16"] = serialization.read_nullable_int16 |
| 137 | context["write_nullable_int32"] = serialization.write_nullable_int32 |
| 138 | context["read_nullable_int32"] = serialization.read_nullable_int32 |
| 139 | context["write_nullable_pyint64"] = serialization.write_nullable_pyint64 |
| 140 | context["read_nullable_pyint64"] = serialization.read_nullable_pyint64 |
| 141 | context["write_nullable_float32"] = serialization.write_nullable_float32 |
| 142 | context["read_nullable_float32"] = serialization.read_nullable_float32 |
| 143 | context["write_nullable_pyfloat64"] = serialization.write_nullable_pyfloat64 |
| 144 | context["read_nullable_pyfloat64"] = serialization.read_nullable_pyfloat64 |
| 145 | context["write_nullable_pystr"] = serialization.write_nullable_pystr |
| 146 | context["read_nullable_pystr"] = serialization.read_nullable_pystr |
| 147 | stmts = [f"{ident(statement)}" for statement in stmts] |
| 148 | # Sanitize the function name to ensure it is valid Python syntax |
| 149 | sanitized_function_name = _sanitize_function_name(function_name) |
| 150 | stmts.insert(0, f"def {sanitized_function_name}({', '.join(params)}):") |
| 151 | stmts = [f"{statement} # line {idx + 1}" for idx, statement in enumerate(stmts)] |
| 152 | code = "\n".join(stmts) |
| 153 | filename = _generate_filename(function_name) |
| 154 | code_dir = _get_code_dir() |
| 155 | if code_dir: |
| 156 | filename = os.path.join(code_dir, filename) |
| 157 | with open(filename, "w") as f: |
| 158 | f.write(code) |
| 159 | f.flush() |
| 160 | if _delete_code_on_exit(): |
| 161 | atexit.register(os.remove, filename) |
| 162 | try: |
| 163 | compiled = compile(code, filename, "exec") |
| 164 | except Exception as e: |
| 165 | raise CompileError(f"Failed to compile code:\n{code}") from e |
| 166 | exec(compiled, context, context) |
| 167 | # See https://stackoverflow.com/questions/64879414/how-does-attrs-fool-the-debugger-to-step-into-auto-generated-code # noqa: E501 |
| 168 | # In order of debuggers like PDB being able to step through the code, |
| 169 | # we add a fake linecache entry. |
| 170 | linecache.cache[filename] = ( |
| 171 | len(code), |
| 172 | None, |
| 173 | code.splitlines(True), |
| 174 | filename, |
| 175 | ) |
| 176 | # Use the sanitized function name to retrieve the function from context |
| 177 | sanitized_function_name = _sanitize_function_name(function_name) |
nothing calls this directly
no test coverage detected