[Serve] Implement flask_request and named python request (#5849)

* Implement flask_request and named python request

* Forgot to include missing files

* Address comment

* Add flask to requirements for doc (lint failed)

* Update doc requirement so lint will build

* Install flask in CI

* Fix typo in .travis.yml

python/ray/experimental/serve/tests/test_task_runner.py

@@ -1,3 +1,4 @@
+import pytest
 import ray
 from ray.experimental.serve.queues import CentralizedQueuesActor
 from ray.experimental.serve.task_runner import (
@@ -6,6 +7,7 @@ from ray.experimental.serve.task_runner import (
     TaskRunnerActor,
     wrap_to_ray_error,
 )
+import ray.experimental.serve.context as context
 
 
 def test_runner_basic():
@@ -17,21 +19,14 @@ def test_runner_basic():
 
 
 def test_runner_wraps_error():
-    def echo(i):
-        return i
-
-    assert wrap_to_ray_error(echo, 2) == 2
-
-    def error(_):
-        return 1 / 0
-
-    assert isinstance(wrap_to_ray_error(error, 1), ray.exceptions.RayTaskError)
+    wrapped = wrap_to_ray_error(Exception())
+    assert isinstance(wrapped, ray.exceptions.RayTaskError)
 
 
 def test_runner_actor(serve_instance):
     q = CentralizedQueuesActor.remote()
 
-    def echo(i):
+    def echo(flask_request, i=None):
         return i
 
     CONSUMER_NAME = "runner"
@@ -46,7 +41,12 @@ def test_runner_actor(serve_instance):
 
     for query in [333, 444, 555]:
         result_token = ray.ObjectID(
-            ray.get(q.enqueue_request.remote(PRODUCER_NAME, query)))
+            ray.get(
+                q.enqueue_request.remote(
+                    PRODUCER_NAME,
+                    request_args=None,
+                    request_kwargs={"i": query},
+                    request_context=context.TaskContext.Python)))
         assert ray.get(result_token) == query
 
 
@@ -60,8 +60,8 @@ def test_ray_serve_mixin(serve_instance):
         def __init__(self, inc):
             self.increment = inc
 
-        def __call__(self, context):
-            return context + self.increment
+        def __call__(self, flask_request, i=None):
+            return i + self.increment
 
     @ray.remote
     class CustomActor(MyAdder, RayServeMixin):
@@ -76,5 +76,38 @@ def test_ray_serve_mixin(serve_instance):
 
     for query in [333, 444, 555]:
         result_token = ray.ObjectID(
-            ray.get(q.enqueue_request.remote(PRODUCER_NAME, query)))
+            ray.get(
+                q.enqueue_request.remote(
+                    PRODUCER_NAME,
+                    request_args=None,
+                    request_kwargs={"i": query},
+                    request_context=context.TaskContext.Python)))
         assert ray.get(result_token) == query + 3
+
+
+def test_task_runner_check_context(serve_instance):
+    q = CentralizedQueuesActor.remote()
+
+    def echo(flask_request, i=None):
+        # Accessing the flask_request without web context should throw.
+        return flask_request.args["i"]
+
+    CONSUMER_NAME = "runner"
+    PRODUCER_NAME = "producer"
+
+    runner = TaskRunnerActor.remote(echo)
+
+    runner._ray_serve_setup.remote(CONSUMER_NAME, q)
+    runner._ray_serve_main_loop.remote(runner)
+
+    q.link.remote(PRODUCER_NAME, CONSUMER_NAME)
+    result_token = ray.ObjectID(
+        ray.get(
+            q.enqueue_request.remote(
+                PRODUCER_NAME,
+                request_args=None,
+                request_kwargs={"i": 42},
+                request_context=context.TaskContext.Python)))
+
+    with pytest.raises(ray.exceptions.RayTaskError):
+        ray.get(result_token)