(plugin_name, plugin: Plugin)
| 52 | |
| 53 | @app.post("/plugins/{plugin_name}") |
| 54 | async def run(plugin_name, plugin: Plugin): |
| 55 | plugin_definition = await get_core_plugin_definition(plugin_name) |
| 56 | try: |
| 57 | validate_samples(plugin.samples_b64, plugin.samples_cloud) |
| 58 | |
| 59 | if plugin.samples_b64: |
| 60 | samples = get_from_samples_b64(plugin.samples_b64[0]) |
| 61 | custom_params = get_custom_params(plugin, plugin.samples_b64[0]) |
| 62 | else: |
| 63 | samples = await get_from_samples_cloud(plugin.samples_cloud[0]) |
| 64 | custom_params = get_custom_params(plugin, plugin.samples_cloud[0]) |
| 65 | # Extract samples |
| 66 | |
| 67 | plugin_instance = plugin_definition(**custom_params) # a way to provide params as a single dict |
| 68 | |
| 69 | return plugin_instance.run(samples) # all python plugins should have a run method that takes in the samples |
| 70 | |
| 71 | except ValueError as e: |
| 72 | raise fastapi.HTTPException(status_code=400, detail="Invalid parameters: " + str(e)) |
| 73 | except Exception as e: |
| 74 | logging.error(e) |
| 75 | raise fastapi.HTTPException(status_code=500, detail="Unknown error in plugins_api") |
| 76 | |
| 77 | |
| 78 | async def get_core_plugin_definition(plugin_name: str): |
nothing calls this directly
no test coverage detected