Module astrapy.admin.endpoints
Expand source code
# Copyright DataStax, Inc.
#
# Licensed 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.
from __future__ import annotations
import re
from dataclasses import dataclass
from astrapy.settings.defaults import (
API_ENDPOINT_TEMPLATE_ENV_MAP,
)
database_id_matcher = re.compile(
"^[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}$"
)
api_endpoint_parser = re.compile(
r"https://"
r"([0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12})"
r"-"
r"([a-z0-9\-]+)"
r".apps.astra[\-]{0,1}"
r"(dev|test)?"
r".datastax.com"
)
api_endpoint_description = (
"https://<db uuid, 8-4-4-4-12 hex format>-<db region>.apps.astra.datastax.com"
)
generic_api_url_matcher = re.compile(r"^https?:\/\/[a-zA-Z0-9\-.]+(\:[0-9]{1,6}){0,1}$")
generic_api_url_descriptor = "http[s]://<domain name or IP>[:port]"
@dataclass
class ParsedAPIEndpoint:
"""
The results of successfully parsing an Astra DB API endpoint, for internal
by database metadata-related functions.
Attributes:
database_id: e. g. "01234567-89ab-cdef-0123-456789abcdef".
region: a region ID, such as "us-west1".
environment: a label, whose value is one of Environment.PROD,
Environment.DEV or Environment.TEST.
"""
database_id: str
region: str
environment: str
def parse_api_endpoint(api_endpoint: str) -> ParsedAPIEndpoint | None:
"""
Parse an API Endpoint into a ParsedAPIEndpoint structure.
Args:
api_endpoint: a full API endpoint for the Data API.
Returns:
The parsed ParsedAPIEndpoint. If parsing fails, return None.
"""
match = api_endpoint_parser.match(api_endpoint)
if match and match.groups():
d_id, d_re, d_en_x = match.groups()
return ParsedAPIEndpoint(
database_id=d_id,
region=d_re,
environment=d_en_x if d_en_x else "prod",
)
else:
return None
def api_endpoint_parsing_error_message(failing_url: str) -> str:
"""
Format an error message with a suggestion for the expected url format.
"""
return (
f"Cannot parse the supplied API endpoint ({failing_url}). The endpoint "
f'must be in the following form: "{api_endpoint_description}".'
)
def parse_generic_api_url(api_endpoint: str) -> str | None:
"""
Validate a generic API Endpoint string,
such as `http://10.1.1.1:123` or `https://my.domain`.
Args:
api_endpoint: a string supposedly expressing a valid API Endpoint
(not necessarily an Astra DB one).
Returns:
a normalized (stripped) version of the endpoint if valid. If invalid,
return None.
"""
if api_endpoint and api_endpoint[-1] == "/":
_api_endpoint = api_endpoint[:-1]
else:
_api_endpoint = api_endpoint
match = generic_api_url_matcher.match(_api_endpoint)
if match:
return match[0].rstrip("/")
else:
return None
def generic_api_url_parsing_error_message(failing_url: str) -> str:
"""
Format an error message with a suggestion for the expected url format.
"""
return (
f"Cannot parse the supplied API endpoint ({failing_url}). The endpoint "
f'must be in the following form: "{generic_api_url_descriptor}".'
)
def build_api_endpoint(environment: str, database_id: str, region: str) -> str:
"""
Build the API Endpoint full strings from database parameters.
Args:
environment: a label, whose value can be Environment.PROD
or another of Environment.* for which this operation makes sense.
database_id: e. g. "01234567-89ab-cdef-0123-456789abcdef".
region: a region ID, such as "us-west1".
Returns:
the endpoint string, such as "https://01234567-...-eu-west1.apps.datastax.com"
"""
return API_ENDPOINT_TEMPLATE_ENV_MAP[environment].format(
database_id=database_id,
region=region,
)
Functions
def api_endpoint_parsing_error_message(failing_url: str) ‑> str
-
Format an error message with a suggestion for the expected url format.
Expand source code
def api_endpoint_parsing_error_message(failing_url: str) -> str: """ Format an error message with a suggestion for the expected url format. """ return ( f"Cannot parse the supplied API endpoint ({failing_url}). The endpoint " f'must be in the following form: "{api_endpoint_description}".' )
def build_api_endpoint(environment: str, database_id: str, region: str) ‑> str
-
Build the API Endpoint full strings from database parameters.
Args
environment
- a label, whose value can be Environment.PROD or another of Environment.* for which this operation makes sense.
database_id
- e. g. "01234567-89ab-cdef-0123-456789abcdef".
region
- a region ID, such as "us-west1".
Returns
the endpoint string, such as "https://01234567-…-eu-west1.apps.datastax.com"
Expand source code
def build_api_endpoint(environment: str, database_id: str, region: str) -> str: """ Build the API Endpoint full strings from database parameters. Args: environment: a label, whose value can be Environment.PROD or another of Environment.* for which this operation makes sense. database_id: e. g. "01234567-89ab-cdef-0123-456789abcdef". region: a region ID, such as "us-west1". Returns: the endpoint string, such as "https://01234567-...-eu-west1.apps.datastax.com" """ return API_ENDPOINT_TEMPLATE_ENV_MAP[environment].format( database_id=database_id, region=region, )
def generic_api_url_parsing_error_message(failing_url: str) ‑> str
-
Format an error message with a suggestion for the expected url format.
Expand source code
def generic_api_url_parsing_error_message(failing_url: str) -> str: """ Format an error message with a suggestion for the expected url format. """ return ( f"Cannot parse the supplied API endpoint ({failing_url}). The endpoint " f'must be in the following form: "{generic_api_url_descriptor}".' )
def parse_api_endpoint(api_endpoint: str) ‑> ParsedAPIEndpoint | None
-
Parse an API Endpoint into a ParsedAPIEndpoint structure.
Args
api_endpoint
- a full API endpoint for the Data API.
Returns
The parsed ParsedAPIEndpoint. If parsing fails, return None.
Expand source code
def parse_api_endpoint(api_endpoint: str) -> ParsedAPIEndpoint | None: """ Parse an API Endpoint into a ParsedAPIEndpoint structure. Args: api_endpoint: a full API endpoint for the Data API. Returns: The parsed ParsedAPIEndpoint. If parsing fails, return None. """ match = api_endpoint_parser.match(api_endpoint) if match and match.groups(): d_id, d_re, d_en_x = match.groups() return ParsedAPIEndpoint( database_id=d_id, region=d_re, environment=d_en_x if d_en_x else "prod", ) else: return None
def parse_generic_api_url(api_endpoint: str) ‑> str | None
-
Validate a generic API Endpoint string, such as
http://10.1.1.1:123
orhttps://my.domain
.Args
api_endpoint
- a string supposedly expressing a valid API Endpoint
(not necessarily an Astra DB one).
Returns
a normalized (stripped) version of the endpoint if valid. If invalid, return None.
Expand source code
def parse_generic_api_url(api_endpoint: str) -> str | None: """ Validate a generic API Endpoint string, such as `http://10.1.1.1:123` or `https://my.domain`. Args: api_endpoint: a string supposedly expressing a valid API Endpoint (not necessarily an Astra DB one). Returns: a normalized (stripped) version of the endpoint if valid. If invalid, return None. """ if api_endpoint and api_endpoint[-1] == "/": _api_endpoint = api_endpoint[:-1] else: _api_endpoint = api_endpoint match = generic_api_url_matcher.match(_api_endpoint) if match: return match[0].rstrip("/") else: return None
Classes
class ParsedAPIEndpoint (database_id: str, region: str, environment: str)
-
The results of successfully parsing an Astra DB API endpoint, for internal by database metadata-related functions.
Attributes
database_id
- e. g. "01234567-89ab-cdef-0123-456789abcdef".
region
- a region ID, such as "us-west1".
environment
- a label, whose value is one of Environment.PROD, Environment.DEV or Environment.TEST.
Expand source code
@dataclass class ParsedAPIEndpoint: """ The results of successfully parsing an Astra DB API endpoint, for internal by database metadata-related functions. Attributes: database_id: e. g. "01234567-89ab-cdef-0123-456789abcdef". region: a region ID, such as "us-west1". environment: a label, whose value is one of Environment.PROD, Environment.DEV or Environment.TEST. """ database_id: str region: str environment: str
Class variables
var database_id : str
var environment : str
var region : str