Remove asyncio

croud/api.py

@@ -0,0 +1,152 @@
+# Licensed to CRATE Technology GmbH ("Crate") under one or more contributor
+# license agreements.  See the NOTICE file distributed with this work for
+# additional information regarding copyright ownership.  Crate licenses
+# this file to you under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.  You may
+# obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.  See the
+# License for the specific language governing permissions and limitations
+# under the License.
+#
+# However, if you have executed another commercial license agreement
+# with Crate these terms will supersede the license and you may use the
+# software solely pursuant to the terms of the relevant commercial agreement.
+
+import enum
+from argparse import Namespace
+from typing import Dict, Optional, Tuple
+
+import requests
+from yarl import URL
+
+from croud.config import Configuration
+from croud.printer import print_error
+
+ResponsePair = Tuple[Optional[Dict], Optional[Dict]]
+
+CLOUD_LOCAL_URL = "http://localhost:8000"
+CLOUD_DEV_DOMAIN = "cratedb-dev.cloud"
+CLOUD_PROD_DOMAIN = "cratedb.cloud"
+
+
+class RequestMethod(enum.Enum):
+    DELETE = "delete"
+    GET = "get"
+    PATCH = "patch"
+    POST = "post"
+    PUT = "put"
+
+
+class Client:
+    def __init__(self, *, env: str, region: str, sudo: bool = False):
+        self.env = env or Configuration.get_env()
+        self.region = region or Configuration.get_setting("region")
+        self.sudo = sudo
+
+        self.base_url = URL(cloud_url(self.env, self.region))
+
+        self._token = Configuration.get_token(self.env)
+        self.session = requests.Session()
+        if get_verify_ssl() is False:
+            self.session.verify = False
+        self.session.cookies["session"] = self._token
+        if self.sudo:
+            self.session.headers["X-Auth-Sudo"] = "1"
+
+    @staticmethod
+    def from_args(args: Namespace) -> "Client":
+        return Client(env=args.env, region=args.region, sudo=args.sudo)
+
+    def request(
+        self,
+        method: RequestMethod,
+        endpoint: str,
+        *,
+        params: dict = None,
+        body: dict = None,
+    ):
+        kwargs: dict = {"allow_redirects": False}
+        if params is not None:
+            kwargs["params"] = params
+        if body is not None:
+            kwargs["json"] = body
+
+        try:
+            response = self.session.request(
+                method.value, str(self.base_url.with_path(endpoint)), **kwargs
+            )
+        except requests.RequestException as e:
+            message = (
+                f"Failed to perform command on {e.request.url}. "
+                f"Original error was: '{e}' "
+                f"Does the environment exist in the region you specified?"
+            )
+            return None, {"message": message, "success": False}
+
+        if response.is_redirect:  # login redirect
+            print_error("Unauthorized. Use `croud login` to login to CrateDB Cloud.")
+            exit(1)
+
+        # Refresh a previously provided token because it has timed out
+        response_token = response.cookies.get("session")
+        if response_token and response_token != self._token:
+            self._token = response_token
+            Configuration.set_token(response_token, self.env)
+
+        return self.decode_response(response)
+
+    def delete(
+        self, endpoint: str, *, params: dict = None, body: dict = None
+    ) -> ResponsePair:
+        return self.request(RequestMethod.DELETE, endpoint, params=params, body=body)
+
+    def get(self, endpoint: str, *, params: dict = None) -> ResponsePair:
+        return self.request(RequestMethod.GET, endpoint, params=params)
+
+    def patch(
+        self, endpoint: str, *, params: dict = None, body: dict = None
+    ) -> ResponsePair:
+        return self.request(RequestMethod.PATCH, endpoint, params=params, body=body)
+
+    def post(
+        self, endpoint: str, *, params: dict = None, body: dict = None
+    ) -> ResponsePair:
+        return self.request(RequestMethod.POST, endpoint, params=params, body=body)
+
+    def put(
+        self, endpoint: str, *, params: dict = None, body: dict = None
+    ) -> ResponsePair:
+        return self.request(RequestMethod.PUT, endpoint, params=params, body=body)
+
+    def decode_response(self, resp: requests.Response) -> ResponsePair:
+        if resp.status_code == 204:
+            # response is empty
+            return None, None
+
+        try:
+            # API always returns JSON, unless there's an unhandled server error
+            body = resp.json()
+        except ValueError:
+            body = {"message": "Invalid response type.", "success": False}
+
+        if resp.status_code >= 400:
+            return None, body
+        else:
+            return body, None
+
+
+def cloud_url(env: str, region: str = "bregenz.a1") -> str:
+    if env == "local":
+        return CLOUD_LOCAL_URL
+
+    host = CLOUD_DEV_DOMAIN if env == "dev" else CLOUD_PROD_DOMAIN
+    return f"https://{region}.{host}"
+
+
+def get_verify_ssl() -> bool:
+    return True

croud/clusters/commands.py

@@ -19,10 +19,9 @@
 
 from argparse import Namespace
 
+from croud.api import Client
 from croud.config import get_output_format
 from croud.printer import print_response
-from croud.rest import Client
-from croud.session import RequestMethod
 from croud.util import require_confirmation
 
 
@@ -36,7 +35,7 @@ def clusters_list(args: Namespace) -> None:
         params["project_id"] = args.project_id
 
     client = Client.from_args(args)
-    data, errors = client.send(RequestMethod.GET, "/api/v2/clusters/", params=params)
+    data, errors = client.get("/api/v2/clusters/", params=params)
     print_response(
         data=data,
         errors=errors,
@@ -72,7 +71,7 @@ def clusters_deploy(args: Namespace) -> None:
     if args.unit:
         body["product_unit"] = args.unit
     client = Client.from_args(args)
-    data, errors = client.send(RequestMethod.POST, "/api/v2/clusters/", body=body)
+    data, errors = client.post("/api/v2/clusters/", body=body)
     print_response(
         data=data,
         errors=errors,
@@ -91,9 +90,7 @@ def clusters_scale(args: Namespace) -> None:
 
     body = {"product_unit": args.unit}
     client = Client.from_args(args)
-    data, errors = client.send(
-        RequestMethod.PUT, f"/api/v2/clusters/{args.cluster_id}/scale/", body=body
-    )
+    data, errors = client.put(f"/api/v2/clusters/{args.cluster_id}/scale/", body=body)
     print_response(
         data=data,
         errors=errors,
@@ -112,9 +109,7 @@ def clusters_upgrade(args: Namespace) -> None:
 
     body = {"crate_version": args.version}
     client = Client.from_args(args)
-    data, errors = client.send(
-        RequestMethod.PUT, f"/api/v2/clusters/{args.cluster_id}/upgrade/", body=body
-    )
+    data, errors = client.put(f"/api/v2/clusters/{args.cluster_id}/upgrade/", body=body)
     print_response(
         data=data,
         errors=errors,
@@ -132,9 +127,7 @@ def clusters_upgrade(args: Namespace) -> None:
 )
 def clusters_delete(args: Namespace) -> None:
     client = Client.from_args(args)
-    data, errors = client.send(
-        RequestMethod.DELETE, f"/api/v2/clusters/{args.cluster_id}/"
-    )
+    data, errors = client.delete(f"/api/v2/clusters/{args.cluster_id}/")
     print_response(
         data=data,
         errors=errors,

croud/consumers/commands.py

@@ -19,10 +19,9 @@
 
 from argparse import Namespace
 
+from croud.api import Client
 from croud.config import get_output_format
 from croud.printer import print_response
-from croud.rest import Client
-from croud.session import RequestMethod
 from croud.util import require_confirmation
 
 
@@ -44,7 +43,7 @@ def consumers_deploy(args: Namespace) -> None:
         "table_schema": args.consumer_schema,
     }
     client = Client.from_args(args)
-    data, errors = client.send(RequestMethod.POST, "/api/v2/consumers/", body=body)
+    data, errors = client.post("/api/v2/consumers/", body=body)
     print_response(
         data=data,
         errors=errors,
@@ -76,7 +75,7 @@ def consumers_list(args: Namespace) -> None:
         params["project_id"] = args.project_id
 
     client = Client.from_args(args)
-    data, errors = client.send(RequestMethod.GET, "/api/v2/consumers/", params=params)
+    data, errors = client.get("/api/v2/consumers/", params=params)
     print_response(
         data=data,
         errors=errors,
@@ -114,9 +113,7 @@ def consumers_edit(args: Namespace) -> None:
     if config:
         body["config"] = config
     client = Client.from_args(args)
-    data, errors = client.send(
-        RequestMethod.PATCH, f"/api/v2/consumers/{args.consumer_id}/", body=body
-    )
+    data, errors = client.patch(f"/api/v2/consumers/{args.consumer_id}/", body=body)
     print_response(
         data=data, errors=errors, keys=["id"], output_fmt=get_output_format(args)
     )
@@ -128,9 +125,7 @@ def consumers_edit(args: Namespace) -> None:
 )
 def consumers_delete(args: Namespace) -> None:
     client = Client.from_args(args)
-    data, errors = client.send(
-        RequestMethod.DELETE, f"/api/v2/consumers/{args.consumer_id}/"
-    )
+    data, errors = client.delete(f"/api/v2/consumers/{args.consumer_id}/")
     print_response(
         data=data,
         errors=errors,

croud/login.py

@@ -17,16 +17,14 @@
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
 
-import asyncio
 from argparse import Namespace
 from functools import partial
 from typing import Optional
 
+from croud.api import Client, cloud_url
 from croud.config import Configuration
-from croud.printer import print_error, print_info
-from croud.rest import Client, RequestMethod
+from croud.printer import print_error, print_info, print_warning
 from croud.server import Server
-from croud.session import cloud_url
 from croud.util import can_launch_browser, open_page_in_browser
 
 LOGIN_PATH = "/oauth2/login?cli=true"
@@ -36,7 +34,7 @@ def get_org_id() -> Optional[str]:
     client = Client(
         env=Configuration.get_env(), region=Configuration.get_setting("region")
     )
-    data, error = client.send(RequestMethod.GET, "/api/v2/users/me/")
+    data, error = client.get("/api/v2/users/me/")
     if data and not error:
         return data.get("organization_id")
     return None
@@ -47,41 +45,31 @@ def login(args: Namespace) -> None:
     Performs an OAuth2 Login to CrateDB Cloud
     """
 
-    if can_launch_browser():
-        env = args.env or Configuration.get_env()
-
-        loop = asyncio.get_event_loop()
-        server = Server(loop)
-        server.create_web_app(partial(Configuration.set_token, env=env))
-        loop.run_until_complete(server.start())
-
-        open_page_in_browser(_login_url(env))
-        print_info("A browser tab has been launched for you to login.")
-
-        try:
-            loop.run_forever()
-        except KeyboardInterrupt:
-            loop.run_until_complete(server.stop())
-            exit(1)
-        finally:
-            loop.run_until_complete(server.stop())
-
-        Configuration.set_context(env.lower())
+    if not can_launch_browser():
+        print_error("Login only works with a valid browser installed.")
+        exit(1)
 
+    env = args.env or Configuration.get_env()
+    server_thread = Server(partial(Configuration.set_token, env=env))
+    Configuration.set_context(env.lower())
+    server_thread.start()
+    open_page_in_browser(login_url(env))
+    print_info("A browser tab has been launched for you to login.")
+    try:
+        # Wait for the user to login. They'll be redirected to the `SetTokenHandler`
+        # which will set the token in the configuration.
+        server_thread.wait()
+    except (KeyboardInterrupt, SystemExit):
+        print_warning("Login cancelled.")
+    else:
         organization_id = get_org_id()
         if organization_id:
             Configuration.set_organization_id(organization_id, env)
         else:
             Configuration.set_organization_id("", env)
 
-        loop.close()
-
-    else:
-        print_error("Login only works with a valid browser installed.")
-        exit(1)
-
-    print_info("Login successful.")
+        print_info("Login successful.")
 
 
-def _login_url(env: str) -> str:
+def login_url(env: str) -> str:
     return cloud_url(env.lower()) + LOGIN_PATH

croud/logout.py

@@ -17,32 +17,25 @@
 # with Crate these terms will supersede the license and you may use the
 # software solely pursuant to the terms of the relevant commercial agreement.
 
-import asyncio
 from argparse import Namespace
 
+from croud.api import Client, cloud_url
 from croud.config import Configuration
 from croud.printer import print_info
-from croud.session import HttpSession, cloud_url
 
 LOGOUT_PATH = "/oauth2/logout"
 
 
 def logout(args: Namespace) -> None:
-    loop = asyncio.get_event_loop()
-    env = args.env or Configuration.get_env()
-    token = Configuration.get_token(env)
+    client = Client.from_args(args)
+    env = client.env
+    client.get(logout_url(env))
 
-    loop.run_until_complete(make_request(env, token))
     Configuration.set_token("", env)
     Configuration.set_organization_id("", env)
 
     print_info("You have been logged out.")
 
 
-async def make_request(env: str, token: str) -> None:
-    async with HttpSession(env, token) as session:
-        await session.logout(_logout_url(env))
-
-
-def _logout_url(env: str) -> str:
+def logout_url(env: str) -> str:
     return cloud_url(env) + LOGOUT_PATH