Skip to content

Commit

Permalink
logging, create_database signature (#271)
Browse files Browse the repository at this point in the history
  • Loading branch information
@hemidactylus
hemidactylus authored Mar 29, 2024
1 parent 6717f25 commit d27273c
Show file tree
Hide file tree
Showing 10 changed files with 373 additions and 58 deletions.
75 changes: 62 additions & 13 deletions astrapy/admin.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from __future__ import annotations

import logging
import re
import time
from abc import ABC, abstractmethod
Expand All @@ -38,8 +39,9 @@
from astrapy import AsyncDatabase, Database


DEFAULT_NEW_DATABASE_CLOUD_PROVIDER = "gcp"
DEFAULT_NEW_DATABASE_REGION = "us-east1"
logger = logging.getLogger(__name__)


DATABASE_POLL_NAMESPACE_SLEEP_TIME = 2
DATABASE_POLL_SLEEP_TIME = 15

Expand Down Expand Up @@ -425,6 +427,7 @@ def set_caller(
... )
"""

logger.info(f"setting caller to {caller_name}/{caller_version}")
self._caller_name = caller_name
self._caller_version = caller_version
self._astra_db_ops.set_caller(caller_name, caller_version)
Expand Down Expand Up @@ -458,9 +461,11 @@ def list_databases(
'eu-west-1'
"""

logger.info("getting databases")
gd_list_response = self._astra_db_ops.get_databases(
timeout_info=base_timeout_info(max_time_ms)
)
logger.info("finished getting databases")
if not isinstance(gd_list_response, list):
raise DevOpsAPIException(
"Faulty response from get-databases DevOps API command.",
Expand Down Expand Up @@ -503,10 +508,12 @@ def database_info(
'eu-west-1'
"""

logger.info(f"getting database info for {id}")
gd_response = self._astra_db_ops.get_database(
database=id,
timeout_info=base_timeout_info(max_time_ms),
)
logger.info(f"finished getting database info for {id}")
if not isinstance(gd_response, dict):
raise DevOpsAPIException(
"Faulty response from get-database DevOps API command.",
Expand All @@ -522,23 +529,26 @@ def create_database(
self,
name: str,
*,
cloud_provider: str,
region: str,
namespace: Optional[str] = None,
wait_until_active: bool = True,
cloud_provider: str = DEFAULT_NEW_DATABASE_CLOUD_PROVIDER,
region: str = DEFAULT_NEW_DATABASE_REGION,
max_time_ms: Optional[int] = None,
) -> AstraDBDatabaseAdmin:
"""
Create a database as requested, optionally waiting for it to be ready.
Args:
name: the desired name for the database.
namespace: name for the one namespace the database starts with.
If omitted, DevOps API will use its default.
cloud_provider: one of 'aws', 'gcp' or 'azure'.
region: any of the available cloud regions.
wait_until_active: if True (default), the method returns only after
the newly-created database is in ACTIVE state (a few minutes,
usually). If False, it will return right after issuing the
creation request to the DevOps API, and it will be responsibility
of the caller to check the database status before working with it.
cloud_provider: one of 'aws', 'gcp' (default) or 'azure'.
region: any of the available cloud regions (default: 'us-east1').
max_time_ms: a timeout, in milliseconds, for the whole requested
operation to complete.
Note that a timeout is no guarantee that the creation request
Expand All @@ -559,25 +569,36 @@ def create_database(
"""

database_definition = {
"name": name,
"tier": "serverless",
"cloudProvider": cloud_provider,
"region": region,
"capacityUnits": 1,
"dbType": "vector",
k: v
for k, v in {
"name": name,
"tier": "serverless",
"cloudProvider": cloud_provider,
"region": region,
"capacityUnits": 1,
"dbType": "vector",
"keyspace": namespace,
}.items()
if v is not None
}
timeout_manager = MultiCallTimeoutManager(
overall_max_time_ms=max_time_ms, exception_type="devops_api"
)
logger.info(f"creating database f{name}/(f{cloud_provider}, f{region})")
cd_response = self._astra_db_ops.create_database(
database_definition=database_definition,
timeout_info=base_timeout_info(max_time_ms),
)
logger.info(
"devops api returned from creating database "
f"f{name}/(f{cloud_provider}, f{region})"
)
if cd_response is not None and "id" in cd_response:
new_database_id = cd_response["id"]
if wait_until_active:
last_status_seen = STATUS_PENDING
while last_status_seen in {STATUS_PENDING, STATUS_INITIALIZING}:
logger.info(f"sleeping to poll for status of '{new_database_id}'")
time.sleep(DATABASE_POLL_SLEEP_TIME)
last_status_seen = self.database_info(
id=new_database_id,
Expand All @@ -588,6 +609,10 @@ def create_database(
f"Database {name} entered unexpected status {last_status_seen} after PENDING"
)
# return the database instance
logger.info(
f"finished creating database '{new_database_id}' = "
f"f{name}/(f{cloud_provider}, f{region})"
)
return AstraDBDatabaseAdmin.from_astra_db_admin(
id=new_database_id,
astra_db_admin=self,
Expand Down Expand Up @@ -638,15 +663,18 @@ def drop_database(
timeout_manager = MultiCallTimeoutManager(
overall_max_time_ms=max_time_ms, exception_type="devops_api"
)
logger.info(f"dropping database '{id}'")
te_response = self._astra_db_ops.terminate_database(
database=id,
timeout_info=base_timeout_info(max_time_ms),
)
logger.info(f"devops api returned from dropping database '{id}'")
if te_response == id:
if wait_until_active:
last_status_seen: Optional[str] = STATUS_TERMINATING
_db_name: Optional[str] = None
while last_status_seen == STATUS_TERMINATING:
logger.info(f"sleeping to poll for status of '{id}'")
time.sleep(DATABASE_POLL_SLEEP_TIME)
#
detected_databases = [
Expand All @@ -666,6 +694,7 @@ def drop_database(
raise DevOpsAPIException(
f"Database {id}{_name_desc} entered unexpected status {last_status_seen} after PENDING"
)
logger.info(f"finished dropping database '{id}'")
return {"ok": 1}
else:
raise DevOpsAPIException(
Expand Down Expand Up @@ -1022,6 +1051,7 @@ def set_caller(
... )
"""

logger.info(f"setting caller to {caller_name}/{caller_version}")
self._astra_db_admin.set_caller(caller_name, caller_version)

@staticmethod
Expand Down Expand Up @@ -1144,10 +1174,13 @@ def info(self, *, max_time_ms: Optional[int] = None) -> AdminDatabaseInfo:
'us-east1'
"""

return self._astra_db_admin.database_info( # type: ignore[no-any-return]
logger.info(f"getting info ('{self.id}')")
req_response = self._astra_db_admin.database_info(
id=self.id,
max_time_ms=max_time_ms,
)
logger.info(f"finished getting info ('{self.id}')")
return req_response # type: ignore[no-any-return]

def list_namespaces(self, *, max_time_ms: Optional[int] = None) -> List[str]:
"""
Expand All @@ -1164,7 +1197,9 @@ def list_namespaces(self, *, max_time_ms: Optional[int] = None) -> List[str]:
['default_keyspace', 'staging_namespace']
"""

logger.info(f"getting namespaces ('{self.id}')")
info = self.info(max_time_ms=max_time_ms)
logger.info(f"finished getting namespaces ('{self.id}')")
if info.raw_info is None:
raise DevOpsAPIException("Could not get the namespace list.")
else:
Expand Down Expand Up @@ -1213,15 +1248,20 @@ def create_namespace(
timeout_manager = MultiCallTimeoutManager(
overall_max_time_ms=max_time_ms, exception_type="devops_api"
)
logger.info(f"creating namespace '{name}' on '{self.id}'")
cn_response = self._astra_db_admin._astra_db_ops.create_keyspace(
database=self.id,
keyspace=name,
timeout_info=base_timeout_info(max_time_ms),
)
logger.info(
f"devops api returned from creating namespace '{name}' on '{self.id}'"
)
if cn_response is not None and name == cn_response.get("name"):
if wait_until_active:
last_status_seen = STATUS_MAINTENANCE
while last_status_seen == STATUS_MAINTENANCE:
logger.info(f"sleeping to poll for status of '{self.id}'")
time.sleep(DATABASE_POLL_NAMESPACE_SLEEP_TIME)
last_status_seen = self.info(
max_time_ms=timeout_manager.remaining_timeout_ms(),
Expand All @@ -1233,6 +1273,7 @@ def create_namespace(
# is the namespace found?
if name not in self.list_namespaces():
raise DevOpsAPIException("Could not create the namespace.")
logger.info(f"finished creating namespace '{name}' on '{self.id}'")
return {"ok": 1}
else:
raise DevOpsAPIException(
Expand Down Expand Up @@ -1281,15 +1322,20 @@ def drop_namespace(
timeout_manager = MultiCallTimeoutManager(
overall_max_time_ms=max_time_ms, exception_type="devops_api"
)
logger.info(f"dropping namespace '{name}' on '{self.id}'")
dk_response = self._astra_db_admin._astra_db_ops.delete_keyspace(
database=self.id,
keyspace=name,
timeout_info=base_timeout_info(max_time_ms),
)
logger.info(
f"devops api returned from dropping namespace '{name}' on '{self.id}'"
)
if dk_response == name:
if wait_until_active:
last_status_seen = STATUS_MAINTENANCE
while last_status_seen == STATUS_MAINTENANCE:
logger.info(f"sleeping to poll for status of '{self.id}'")
time.sleep(DATABASE_POLL_NAMESPACE_SLEEP_TIME)
last_status_seen = self.info(
max_time_ms=timeout_manager.remaining_timeout_ms(),
Expand All @@ -1301,6 +1347,7 @@ def drop_namespace(
# is the namespace found?
if name in self.list_namespaces():
raise DevOpsAPIException("Could not drop the namespace.")
logger.info(f"finished dropping namespace '{name}' on '{self.id}'")
return {"ok": 1}
else:
raise DevOpsAPIException(
Expand Down Expand Up @@ -1350,11 +1397,13 @@ def drop(
which avoids using a deceased database any further.
"""

logger.info(f"dropping this database ('{self.id}')")
return self._astra_db_admin.drop_database( # type: ignore[no-any-return]
id=self.id,
wait_until_active=wait_until_active,
max_time_ms=max_time_ms,
)
logger.info(f"finished dropping this database ('{self.id}')")

def get_database(
self,
Expand Down
7 changes: 7 additions & 0 deletions astrapy/client.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@

from __future__ import annotations

import logging
import re
from typing import Any, Dict, Optional, TYPE_CHECKING

Expand All @@ -32,6 +33,9 @@
from astrapy.admin import AstraDBAdmin


logger = logging.getLogger(__name__)


class DataAPIClient:
"""
A client for using the Data API. This is the main entry point and sits
Expand Down Expand Up @@ -180,6 +184,7 @@ def set_caller(
>>> my_client.set_caller(caller_name="the_caller", caller_version="0.1.0")
"""

logger.info(f"setting caller to {caller_name}/{caller_version}")
self._caller_name = caller_name
self._caller_version = caller_version

Expand Down Expand Up @@ -243,12 +248,14 @@ def get_database(
_region = region
else:
if this_db_info is None:
logger.info(f"fetching raw database info for {id}")
this_db_info = fetch_raw_database_info_from_id_token(
id=id,
token=self.token,
environment=self.environment,
max_time_ms=max_time_ms,
)
logger.info(f"finished fetching raw database info for {id}")
_region = this_db_info["info"]["region"]

_token = token or self.token
Expand Down
Loading

0 comments on commit d27273c

Please sign in to comment.