[serve] Add initial support for @serve.deployment syntax
diff --git a/python/ray/serve/__init__.py b/python/ray/serve/__init__.py
index c39a81c4fcb84..9ace05d219dfa 100644
--- a/python/ray/serve/__init__.py
+++ b/python/ray/serve/__init__.py
@@ -2,7 +2,7 @@
     accept_batch, connect, start, get_replica_context, get_handle,
     shadow_traffic, set_traffic, delete_backend, list_backends, create_backend,
     get_backend_config, update_backend_config, list_endpoints, delete_endpoint,
-    create_endpoint, shutdown, ingress)
+    create_endpoint, shutdown, ingress, deployment)
 from ray.serve.batching import batch
 from ray.serve.config import BackendConfig, HTTPOptions
 from ray.serve.utils import ServeRequest
@@ -18,5 +18,5 @@
     "shadow_traffic", "set_traffic", "delete_backend", "list_backends",
     "create_backend", "get_backend_config", "update_backend_config",
     "list_endpoints", "delete_endpoint", "create_endpoint", "shutdown",
-    "ingress"
+    "ingress", "deployment"
 ]
diff --git a/python/ray/serve/api.py b/python/ray/serve/api.py
index 81f9fd2e59fb7..0a2ed512a0369 100644
--- a/python/ray/serve/api.py
+++ b/python/ray/serve/api.py
@@ -1,3 +1,4 @@
+from abc import ABC
 import asyncio
 import atexit
 import inspect
@@ -1062,3 +1063,129 @@ def decorator(cls):
         return cls
 
     return decorator
+
+
+class ServeDeployment(ABC):
+    @classmethod
+    def deploy(self, *init_args) -> None:
+        """Deploy this deployment.
+
+        Args:
+            *init_args (optional): the arguments to pass to the class __init__
+                method. Not valid if this deployment wraps a function.
+        """
+        # TODO(edoakes): how to avoid copy-pasting the docstrings here?
+        raise NotImplementedError()
+
+    @classmethod
+    def delete(self) -> None:
+        """Delete this deployment."""
+        raise NotImplementedError()
+
+    @classmethod
+    def get_handle(self, sync: Optional[bool] = True
+                   ) -> Union[RayServeHandle, RayServeSyncHandle]:
+        raise NotImplementedError()
+
+
+def make_deployment_cls(
+        backend_def: Callable,
+        name: str,
+        version: str,
+        ray_actor_options: Optional[Dict] = None,
+        config: Optional[BackendConfig] = None) -> ServeDeployment:
+    class Deployment(ServeDeployment):
+        _backend_def = backend_def
+        _name = name
+        _version = version
+        _ray_actor_options = ray_actor_options
+        _config = config
+
+        @classmethod
+        def deploy(self, *init_args):
+            """Deploy this deployment.
+
+            Args:
+                *init_args (optional): args to pass to the class __init__
+                    method. Not valid if this deployment wraps a function.
+            """
+            return _get_global_client().deploy(
+                Deployment._name,
+                Deployment._backend_def,
+                *init_args,
+                ray_actor_options=Deployment._ray_actor_options,
+                config=Deployment._config,
+                version=Deployment._version,
+                _internal=True)
+
+        @classmethod
+        def delete(self):
+            """Delete this deployment."""
+            raise NotImplementedError()
+
+        @classmethod
+        def get_handle(self, sync: Optional[bool] = True
+                       ) -> Union[RayServeHandle, RayServeSyncHandle]:
+            return _get_global_client().get_handle(
+                Deployment._name, missing_ok=False, sync=sync, _internal=True)
+
+        @classmethod
+        def options(self,
+                    backend_def: Optional[Callable] = None,
+                    name: Optional[str] = None,
+                    version: Optional[str] = None,
+                    ray_actor_options: Optional[Dict] = None,
+                    config: Optional[BackendConfig] = None) -> "Deployment":
+            """Return a new deployment with the specified options set."""
+            return make_deployment_cls(
+                backend_def or Deployment._backend_def,
+                name or Deployment._name,
+                version or Deployment._version,
+                ray_actor_options=ray_actor_options
+                or Deployment._ray_actor_options,
+                config=config or Deployment._config,
+            )
+
+    return Deployment
+
+
+# TODO(edoakes): better typing on the return value of the decorator.
+def deployment(name: str,
+               version: Optional[str] = None,
+               ray_actor_options: Optional[Dict] = None,
+               config: Optional[Union[BackendConfig, Dict[str, Any]]] = None
+               ) -> Callable[[Callable], ServeDeployment]:
+    """Define a Serve deployment.
+
+    Args:
+        name (str): Globally-unique name identifying this deployment.
+        version (str): Version of the deployment. This is used to indicate a
+            code change for the deployment; when it is re-deployed with a
+            version change, a rolling update of the replicas will be performed.
+        ray_actor_options (dict): Options to be passed to the Ray actor
+            constructor such as resource requirements.
+        config (dict, serve.BackendConfig, optional): Configuration options
+            for this backend. Either a BackendConfig, or a dictionary
+            mapping strings to values for the following supported options:
+            - "num_replicas": number of processes to start up that
+            will handle requests to this backend.
+            - "max_concurrent_queries": the maximum number of queries that
+            will be sent to a replica of this backend without receiving a
+            response.
+            - "user_config" (experimental): Arguments to pass to the
+            reconfigure method of the backend. The reconfigure method is
+            called if "user_config" is not None.
+
+    Example:
+    >>> @serve.deployment("deployment1", version="v1")
+        class MyDeployment:
+            pass
+
+    >>> MyDeployment.deploy(*init_args)
+    """
+
+    def decorator(backend_def):
+        return make_deployment_cls(backend_def, name, version,
+                                   ray_actor_options, config)
+
+    return decorator
diff --git a/python/ray/serve/tests/test_deploy.py b/python/ray/serve/tests/test_deploy.py
index 20c70b7b77e08..30862354ba98e 100644
--- a/python/ray/serve/tests/test_deploy.py
+++ b/python/ray/serve/tests/test_deploy.py
@@ -4,53 +4,54 @@
 import requests
 
 import ray
+from ray import serve
 
 
 @pytest.mark.parametrize("use_handle", [True, False])
 def test_deploy(serve_instance, use_handle):
-    client = serve_instance
-
     name = "test"
 
+    @serve.deployment(name, version="1")
+    def d(*args):
+        return f"1|{os.getpid()}"
+
     def call():
         if use_handle:
-            ret = ray.get(client.get_handle(name).remote())
+            ret = ray.get(d.get_handle().remote())
         else:
             ret = requests.get(f"http://localhost:8000/{name}").text
 
         return ret.split("|")[0], ret.split("|")[1]
 
-    def v1(*args):
-        return f"1|{os.getpid()}"
-
-    def v2(*args):
-        return f"2|{os.getpid()}"
-
-    client.deploy(name, v1, version="1")
+    d.deploy()
     val1, pid1 = call()
     assert val1 == "1"
 
     # Redeploying with the same version and code should do nothing.
-    client.deploy(name, v1, version="1")
+    d.deploy()
     val2, pid2 = call()
     assert val2 == "1"
     assert pid2 == pid1
 
     # Redeploying with a new version should start a new actor.
-    client.deploy(name, v1, version="2")
+    d.options(version="2").deploy()
     val3, pid3 = call()
     assert val3 == "1"
     assert pid3 != pid2
 
+    @serve.deployment(name, version="2")
+    def d(*args):
+        return f"2|{os.getpid()}"
+
     # Redeploying with the same version and new code should do nothing.
-    client.deploy(name, v2, version="2")
+    d.deploy()
     val4, pid4 = call()
     assert val4 == "1"
     assert pid4 == pid3
 
     # Redeploying with new code and a new version should start a new actor
     # running the new code.
-    client.deploy(name, v2, version="3")
+    d.options(version="3").deploy()
     val5, pid5 = call()
     assert val5 == "2"
     assert pid5 != pid4
@@ -58,19 +59,10 @@ def v2(*args):
 
 @pytest.mark.parametrize("use_handle", [True, False])
 def test_config_change(serve_instance, use_handle):
-    client = serve_instance
-
     name = "test"
 
-    def call():
-        if use_handle:
-            ret = ray.get(client.get_handle(name).remote())
-        else:
-            ret = requests.get(f"http://localhost:8000/{name}").text
-
-        return ret.split("|")[0], ret.split("|")[1]
-
-    class Backend:
+    @serve.deployment(name, version="1")
+    class D:
         def __init__(self):
             self.ret = "1"
 
@@ -80,44 +72,40 @@ def reconfigure(self, d):
         def __call__(self, *args):
             return f"{self.ret}|{os.getpid()}"
 
+    def call():
+        if use_handle:
+            ret = ray.get(D.get_handle().remote())
+        else:
+            ret = requests.get(f"http://localhost:8000/{name}").text
+
+        return ret.split("|")[0], ret.split("|")[1]
+
     # First deploy with no user config set.
-    client.deploy(name, Backend, version="1")
+    D.deploy()
     val1, pid1 = call()
     assert val1 == "1"
 
     # Now update the user config without changing versions. Actor should stay
     # alive but return value should change.
-    client.deploy(
-        name, Backend, version="1", config={"user_config": {
-            "ret": "2"
-        }})
+    D.options(config={"user_config": {"ret": "2"}}).deploy()
     val2, pid2 = call()
     assert pid2 == pid1
     assert val2 == "2"
 
     # Update the user config without changing the version again.
-    client.deploy(
-        name, Backend, version="1", config={"user_config": {
-            "ret": "3"
-        }})
+    D.options(config={"user_config": {"ret": "3"}}).deploy()
     val3, pid3 = call()
     assert pid3 == pid2
     assert val3 == "3"
 
     # Update the version without changing the user config.
-    client.deploy(
-        name, Backend, version="2", config={"user_config": {
-            "ret": "3"
-        }})
+    D.options(version="2", config={"user_config": {"ret": "3"}}).deploy()
     val4, pid4 = call()
     assert pid4 != pid3
     assert val4 == "3"
 
     # Update the version and the user config.
-    client.deploy(
-        name, Backend, version="3", config={"user_config": {
-            "ret": "4"
-        }})
+    D.options(version="3", config={"user_config": {"ret": "4"}}).deploy()
     val5, pid5 = call()
     assert pid5 != pid4
     assert val5 == "4"

<!-- Thank you for your contribution! Please review https://github.com/ray-project/ray/blob/master/CONTRIBUTING.rst before opening a pull request. -->

<!-- Please add a reviewer to the assignee section when you create a PR. If you don't have the access to it, we will shortly find a reviewer and assign them to your PR. -->

## Why are these changes needed?

<!-- Please give a short summary of the change and the problem this solves. -->

## Related issue number

Closes https://github.com/ray-project/ray/milestone/39

## Checks

- [ ] I've run `scripts/format.sh` to lint the changes in this PR.
- [ ] I've included any doc changes needed for https://docs.ray.io/en/master/.
- [ ] I've made sure the tests are passing. Note that there might be a few flaky tests, see the recent failures at https://flakey-tests.ray.io/
- Testing Strategy
   - [ ] Unit tests
   - [ ] Release tests
   - [ ] This PR is not tested :(
