Valentin Baert
05/20/2022, 11:59 AMflow
decorated function "MissingContextError: No settings context found"
What am I missing ?Anna Geller
Valentin Baert
05/20/2022, 12:01 PMfrom flask import request
from flask_api import FlaskAPI
from prefect import flow, get_run_logger
application = FlaskAPI(__name__)
@flow(name="prefect_2_http_server")
def process_request(data):
logger = get_run_logger()
<http://logger.info|logger.info>("new prefect_2_http_server flow : {]".format(data))
@application.route('/event', methods=['POST'])
def push_event():
process_request(request.data)
return {
"ok": True
}
if __name__ == '__main__':
application.run(host='0.0.0.0', debug=True, port=8080)
13:59:23.491 | INFO | werkzeug - 127.0.0.1 - - [20/May/2022 13:59:23] "POST /event HTTP/1.1" 500 -
Traceback (most recent call last):
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 2095, in __call__
return self.wsgi_app(environ, start_response)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 2080, in wsgi_app
response = self.handle_exception(e)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 2077, in wsgi_app
response = self.full_dispatch_request()
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 1525, in full_dispatch_request
rv = self.handle_user_exception(e)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask_api/app.py", line 106, in handle_user_exception
raise e
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 1523, in full_dispatch_request
rv = self.dispatch_request()
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/flask/app.py", line 1509, in dispatch_request
return self.ensure_sync(self.view_functions[rule.endpoint])(**req.view_args)
File "/home/val/dev/prefect-poc/src/flows/prefect_2_http_server_min.py", line 14, in push_event
process_request(request.data)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/prefect/flows.py", line 319, in __call__
return enter_flow_run_engine_from_flow_call(self, parameters)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/prefect/engine.py", line 110, in enter_flow_run_engine_from_flow_call
return anyio.run(begin_run)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/anyio/_core/_eventloop.py", line 70, in run
return asynclib.run(func, *args, **backend_options)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 292, in run
return native_run(wrapper(), debug=debug)
File "/home/val/.pyenv/versions/3.8.12/lib/python3.8/asyncio/runners.py", line 44, in run
return loop.run_until_complete(main)
File "/home/val/.pyenv/versions/3.8.12/lib/python3.8/asyncio/base_events.py", line 616, in run_until_complete
return future.result()
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/anyio/_backends/_asyncio.py", line 287, in wrapper
return await func(*args)
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/prefect/client.py", line 91, in with_injected_client
client_context = get_client()
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/prefect/client.py", line 101, in get_client
ctx = prefect.context.get_settings_context()
File "/home/val/dev/prefect-poc/venv/lib/python3.8/site-packages/prefect/context.py", line 230, in get_settings_context
raise MissingContextError("No settings context found.")
prefect.exceptions.MissingContextError: No settings context found.
Anna Geller
prefect config view
Aaron Goebel
05/20/2022, 7:06 PMAnna Geller
import asyncio
from prefect.client import get_client
async def main():
async with get_client() as client:
depl_id = "074db2e5-229a-460e-85ad-fca31b379fd2"
response = await client.create_flow_run_from_deployment(depl_id)
print(response)
if __name__ == "__main__":
asyncio.run(main())
here are more detailsasync def create_flow_run_from_deployment(
self,
deployment_id: UUID,
*,
parameters: Dict[str, Any] = None,
context: dict = None,
state: schemas.states.State = None,
flow_runner: "FlowRunner" = None,
) -> schemas.core.FlowRun:
"""
Create a flow run for a deployment.
Args:
deployment: The deployment model to create the flow run from
parameters: Parameter overrides for this flow run. Merged with the
deployment defaults
context: Optional run context data
state: The initial state for the run. If not provided, defaults to
`Scheduled` for now. Should always be a `Scheduled` type.
flow_runner: An optional flow runnner to use to execute this flow run.
Raises:
httpx.RequestError: if Orion does not successfully create a run for any reason
Returns:
The flow run model
"""
docstringAaron Goebel
05/20/2022, 7:12 PMAnna Geller
Aaron Goebel
05/20/2022, 7:13 PMAnna Geller
Aaron Goebel
05/20/2022, 7:13 PMAnna Geller
prefect config set PREFECT_API_URL=your_url
Aaron Goebel
05/20/2022, 7:14 PMAnna Geller
@flow
will make the flow run visible in the Orion UIValentin Baert
05/23/2022, 7:14 AMprefect config view
, everything looks right :Anna Geller
Valentin Baert
05/23/2022, 12:06 PMAnna Geller
Valentin Baert
05/23/2022, 2:16 PMAnna Geller
Valentin Baert
05/23/2022, 2:28 PMAnna Geller
Valentin Baert
05/23/2022, 3:05 PMAnna Geller
Zanie
Valentin Baert
05/24/2022, 8:05 AM$ pipdeptree --reverse --packages anyio
Warning!!! Possibly conflicting dependencies found:
* grpcio-status==1.46.1
- grpcio [required: >=1.46.1, installed: 1.43.0]
------------------------------------------------------------------------
anyio==3.6.1
- httpcore==0.14.7 [requires: anyio==3.*]
- httpx==0.22.0 [requires: httpcore>=0.14.5,<0.15.0]
- prefect==2.0b4 [requires: httpx>=0.22]
- prefect==2.0b4 [requires: anyio>=3.4.0]
- starlette==0.19.1 [requires: anyio>=3.4.0,<5]
- fastapi==0.78.0 [requires: starlette==0.19.1]
- prefect==2.0b4 [requires: fastapi>=0.70]
$ pip freeze | grep anyio
anyio==3.6.1
python --version
Python 3.8.12
prefect==2.0b4
Flask-API
Anna Geller
from flask import Flask
from prefect import flow, get_run_logger
app = Flask(__name__)
@flow(name="valentin_flask")
def process_request():
logger = get_run_logger()
<http://logger.info|logger.info>("Hello from Flask - received input: %s")
@app.route("/event", methods=["GET"])
def push_event():
process_request()
return {"ok": True}
if __name__ == "__main__":
app.run(host="0.0.0.0", debug=True, port=8080)
But running a FastAPI app instead 😎
"""
uvicorn valentin_fastapi:app
"""
from fastapi import FastAPI
from prefect import flow, get_run_logger
app = FastAPI(title="Valentin App")
@flow(name="valentin_flask")
def process_request():
logger = get_run_logger()
<http://logger.info|logger.info>("Hello from Flask - received input: %s")
@app.get("/event")
def push_event():
process_request()
return {"ok": True}
works perfectly fine:
INFO: Started server process [34419]
INFO: Waiting for application startup.
INFO: Application startup complete.
INFO: Uvicorn running on <http://127.0.0.1:8000> (Press CTRL+C to quit)
INFO: 127.0.0.1:53096 - "GET / HTTP/1.1" 404 Not Found
INFO: 127.0.0.1:53096 - "GET /favicon.ico HTTP/1.1" 404 Not Found
INFO: 127.0.0.1:53102 - "GET /docs HTTP/1.1" 200 OK
INFO: 127.0.0.1:53102 - "GET /openapi.json HTTP/1.1" 200 OK
14:53:08.146 | INFO | prefect.engine - Created flow run 'qualified-herring' for flow 'valentin_flask'
14:53:08.147 | INFO | Flow run 'qualified-herring' - Using task runner 'ConcurrentTaskRunner'
14:53:08.741 | INFO | Flow run 'qualified-herring' - Hello from Flask - received input: %s
14:53:09.664 | INFO | Flow run 'qualified-herring' - Finished in state Completed()
Valentin Baert
05/24/2022, 12:57 PMZanie
Anna Geller
b-long
08/05/2022, 4:21 PMdocker-compose
stack) and then a separate "Hello World" flask app that invokes a job with GraphQL or some other API?
How do folks normally do this end-to-end?
1. Register the flow via python CLI, results in a flow ID
2. <save the flow ID>
a. How is it extracted from step 1?
3. Setup the Flask project to invoke Prefect, using <saved flow ID>
Thanks in advance!Zanie
Anna Geller