{
  "nbformat": 4,
  "nbformat_minor": 5,
  "metadata": {
    "kernelspec": {
      "display_name": "Python 3",
      "language": "python",
      "name": "python3"
    },
    "language_info": {
      "name": "python",
      "version": "3.13.0"
    },
    "blog_metadata": {
      "topic": "Cosmos DB Shell and the future of AI-assisted database operations",
      "slug": "cosmos-db-shell-and-the-future-of-ai-assisted-database-opera",
      "generated_by": "LinkedIn Post Generator + Azure OpenAI",
      "generated_at": "2026-05-07T14:11:49.789Z"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "# Cosmos DB Shell and the future of AI-assisted database operations\n",
        "\n",
        "Microsoft’s Cosmos DB Shell signals a practical shift toward AI assistance inside day-to-day database workflows. The key value is not autonomous execution, but faster first drafts for queries, troubleshooting, and schema exploration with human review still in the loop. This notebook turns the blog’s ideas into hands-on validation patterns using Python, including deterministic SDK execution, AI-suggested query review, and production guardrail simulations."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "%pip install azure-cosmos python-dotenv"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import os\n",
        "import json\n",
        "from datetime import datetime, timedelta\n",
        "\n",
        "try:\n",
        "    from azure.cosmos import CosmosClient\n",
        "except Exception:\n",
        "    CosmosClient = None"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Deterministic Cosmos DB query workflow with the SDK\n",
        "\n",
        "This example demonstrates a direct, deterministic Cosmos DB query using the Python SDK. It reflects the blog’s point that AI assistance should complement, not replace, standard execution paths engineers can validate and trust."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "### Required environment variables\n",
        "\n",
        "- `COSMOS_URI`: Cosmos DB account URI\n",
        "- `COSMOS_KEY`: Cosmos DB account key\n",
        "- Optional: ensure database `appdb` and container `orders` exist before running against a live environment"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import os\n",
        "\n",
        "try:\n",
        "    from azure.cosmos import CosmosClient\n",
        "except Exception as e:\n",
        "    CosmosClient = None\n",
        "    print(\"azure-cosmos import unavailable:\", e)\n",
        "\n",
        "COSMOS_URI = os.environ.get(\"COSMOS_URI\")\n",
        "COSMOS_KEY = os.environ.get(\"COSMOS_KEY\")\n",
        "DATABASE_NAME = \"appdb\"\n",
        "CONTAINER_NAME = \"orders\"\n",
        "\n",
        "query = \"SELECT TOP 5 c.id, c.status, c.total FROM c WHERE c.status = @status\"\n",
        "params = [{\"name\": \"@status\", \"value\": \"Open\"}]\n",
        "\n",
        "print(\"Prepared query:\")\n",
        "print(query)\n",
        "print(\"Parameters:\", params)\n",
        "\n",
        "if not COSMOS_URI or not COSMOS_KEY:\n",
        "    print(\"Skipping live execution because COSMOS_URI and/or COSMOS_KEY are not set.\")\n",
        "elif CosmosClient is None:\n",
        "    print(\"Skipping live execution because azure-cosmos is not available.\")\n",
        "else:\n",
        "    try:\n",
        "        client = CosmosClient(COSMOS_URI, credential=COSMOS_KEY)\n",
        "        container = client.get_database_client(DATABASE_NAME).get_container_client(CONTAINER_NAME)\n",
        "        results = list(\n",
        "            container.query_items(\n",
        "                query=query,\n",
        "                parameters=params,\n",
        "                enable_cross_partition_query=True,\n",
        "            )\n",
        "        )\n",
        "        print(f\"Returned {len(results)} item(s):\")\n",
        "        for item in results:\n",
        "            print(item)\n",
        "    except Exception as ex:\n",
        "        print(\"Live Cosmos DB query failed:\", ex)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Compare SDK execution with a hypothetical AI-assisted Shell suggestion\n",
        "\n",
        "This example models the conversational workflow described in the blog: a user expresses intent in natural language, an AI system drafts a query, and a human reviews it before execution. The validation step is where production discipline matters most."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "from datetime import datetime, timedelta\n",
        "\n",
        "natural_language = \"Show open orders over $100 from the last 7 days\"\n",
        "seven_days_ago = (datetime.utcnow() - timedelta(days=7)).strftime(\"%Y-%m-%d\")\n",
        "\n",
        "shell_suggested_query = f\"\"\"\n",
        "SELECT c.id, c.total, c.createdAt\n",
        "FROM c\n",
        "WHERE c.status = 'Open'\n",
        "  AND c.total > 100\n",
        "  AND c.createdAt >= '{seven_days_ago}'\n",
        "\"\"\"\n",
        "\n",
        "print(\"Prompt:\", natural_language)\n",
        "print(\"Suggested query:\")\n",
        "print(shell_suggested_query.strip())\n",
        "print(\"Best practice: review, validate partition scope, then execute via SDK or Shell.\")\n",
        "\n",
        "review_checklist = {\n",
        "    \"syntax_reviewed\": True,\n",
        "    \"partition_scope_reviewed\": False,\n",
        "    \"ru_impact_estimated\": False,\n",
        "    \"approved_for_execution\": False,\n",
        "}\n",
        "\n",
        "print(\"\\nReview checklist:\")\n",
        "for k, v in review_checklist.items():\n",
        "    print(f\"- {k}: {v}\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Environment-scoped guardrails before Cosmos DB operations\n",
        "\n",
        "The blog emphasizes governance as the real production test. This Python version of the PowerShell example validates whether a role is allowed to operate in a given environment before any database action proceeds."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "### Required environment variables\n",
        "\n",
        "- `COSMOS_ROLE`: current effective role, such as `reader` or `contributor`"
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "import os\n",
        "\n",
        "def validate_environment_scope(environment=\"dev\", database=\"appdb\", container=\"orders\"):\n",
        "    allowed = {\n",
        "        \"dev\": [\"reader\", \"contributor\"],\n",
        "        \"test\": [\"reader\"],\n",
        "        \"prod\": [\"reader\"],\n",
        "    }\n",
        "\n",
        "    current_role = os.environ.get(\"COSMOS_ROLE\", \"reader\")\n",
        "    if current_role not in allowed[environment]:\n",
        "        raise PermissionError(f\"Role '{current_role}' is not allowed in '{environment}'.\")\n",
        "\n",
        "    return f\"Approved: role={current_role} env={environment} db={database} container={container}\"\n",
        "\n",
        "for env_name in [\"dev\", \"test\", \"prod\"]:\n",
        "    try:\n",
        "        print(validate_environment_scope(environment=env_name))\n",
        "    except Exception as ex:\n",
        "        print(f\"Blocked for {env_name}: {ex}\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Scripted validation pattern for AI-assisted operational commands\n",
        "\n",
        "This example simulates a higher-risk operational request, such as scaling RU/s in production. It shows how AI-generated or AI-assisted actions should still pass change-window checks and approval thresholds before execution."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "task = {\n",
        "    \"Prompt\": \"Scale RU/s for orders container to 10000 in prod\",\n",
        "    \"Environment\": \"prod\",\n",
        "    \"RequestedThroughput\": 10000,\n",
        "    \"ChangeWindowOpen\": False,\n",
        "}\n",
        "\n",
        "print(\"Task:\")\n",
        "print(json.dumps(task, indent=2))\n",
        "\n",
        "try:\n",
        "    if task[\"Environment\"] == \"prod\" and not task[\"ChangeWindowOpen\"]:\n",
        "        raise RuntimeError(\"Blocked: production throughput changes require an open change window.\")\n",
        "\n",
        "    if task[\"RequestedThroughput\"] > 5000:\n",
        "        print(\"Require approval: high RU/s change detected.\")\n",
        "    else:\n",
        "        print(\"Safe to proceed with scripted execution.\")\n",
        "except Exception as ex:\n",
        "    print(ex)"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Workflow model: AI-assisted database operations with human review and policy checks\n",
        "\n",
        "The original post included a Mermaid diagram. Here, the same flow is represented as executable Python so the process can be inspected and adapted in notebook form."
      ]
    },
    {
      "cell_type": "code",
      "metadata": {},
      "source": [
        "workflow_steps = [\n",
        "    \"Engineer asks in Cosmos DB Shell\",\n",
        "    \"AI interprets intent\",\n",
        "    \"Suggests query or operation\",\n",
        "    \"Human reviews scope, cost, and safety\",\n",
        "    \"Policy checks: env, role, approvals\",\n",
        "    \"Execute via Shell or SDK\",\n",
        "    \"Results + audit trail\",\n",
        "]\n",
        "\n",
        "for i, step in enumerate(workflow_steps, start=1):\n",
        "    print(f\"{i}. {step}\")\n",
        "\n",
        "print(\"\\nKey principle: less memorization, more review.\")"
      ],
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {},
      "source": [
        "## Summary\n",
        "\n",
        "Cosmos DB Shell represents a broader move toward AI-assisted database operations becoming part of the expected workflow rather than a novelty. The practical value comes from accelerating first drafts, incident triage, and exploration, while keeping humans responsible for correctness, partition awareness, RU impact, and governance.\n",
        "\n",
        "## Next Steps\n",
        "\n",
        "- Test AI-generated query drafts against known-good SDK queries.\n",
        "- Define approval thresholds for high-impact operations such as throughput changes.\n",
        "- Enforce least-privilege access and environment scoping for all operational paths.\n",
        "- Add audit logging and review checkpoints before production execution.\n",
        "- Evaluate rollout criteria such as query accuracy, workflow design, and production guardrails."
      ]
    }
  ]
}