try:
            self._refresh_schema()
        except NeptuneQueryException:
            raise ValueError("Could not get schema for Neptune database")
    @property
    def get_schema(self) -> str:
        """Returns the schema of the Neptune database"""
        return self.schema
[docs]    def query(self, query: str, params: dict = {}) -> Dict[str, Any]:
        """Query Neptune database."""
        response = requests.post(url=self.query_url, data={"query": query})
        if response.ok:
            results = json.loads(response.content.decode())
            return results
        else:
            raise NeptuneQueryException(
                {
                    "message": "The generated query failed to execute",
                    "details": response.content.decode(),
                }
            )
    def _get_summary(self) -> Dict:
        response = requests.get(url=self.summary_url)
        if not response.ok:
            raise NeptuneQueryException(
                {
                    "message": (
                        "Summary API is not available for this instance of Neptune,"
                        "ensure the engine version is >=1.2.1.0"
                    ),
                    "details": response.content.decode(),
                }
            )
        try:
            summary = response.json()["payload"]["graphSummary"]
        except Exception:
            raise NeptuneQueryException(
                {
                    "message": "Summary API did not return a valid response.",
                    "details": response.content.decode(),
                }
            )
        else:
            return summary
    def _get_labels(self) -> Tuple[List[str], List[str]]:
        """Get node and edge labels from the Neptune statistics summary"""
        summary = self._get_summary()
        n_labels = summary["nodeLabels"]
        e_labels = summary["edgeLabels"]