🚀 AI-Prishtina Microsoft Graph API

The most comprehensive, production-ready Python library for Microsoft Graph API

Async-first • Type-safe • Enterprise-grade • Clean Architecture

Quick Start • Installation • Features • Architecture • API Coverage • Documentation

☕ Support This Project

If you find this project helpful, please consider supporting it:

📋 Table of Contents

• Installation
• Why This Library?
• Quick Start (5 minutes)
• Feature Overview
• Clean Architecture
• Complete API Coverage
• Advanced Features
• Enterprise Features
• Configuration
• Monitoring & Observability
• Competitive Advantages
• Documentation
• Testing
• License & Support

📦 Installation

From PyPI (Recommended)

pip install ai-prishtina-microsoft-graph-api

From Source

git clone https://github.com/ai-prishtina/ai-prishtina-microsoft-graph-api.git
cd ai-prishtina-microsoft-graph-api
pip install -e .

Verify Installation

from ai_prishtina_microsoft_graph_api import GraphAPIClient, __version__
print(f"Installed version: {__version__}")

🎯 Why This Library?

AI-Prishtina Microsoft Graph API is a production-ready, async-first Python library that provides the most comprehensive Microsoft Graph API integration available. Built with Clean Architecture principles, it offers enterprise-grade features that alternatives lack.

Key Differentiators

⚡ Quick Start (5 minutes)

Installation

# Basic installation
pip install ai-prishtina-microsoft-graph-api

# With Redis caching support
pip install ai-prishtina-microsoft-graph-api[redis]

# With all optional dependencies
pip install ai-prishtina-microsoft-graph-api[all]

Minimal Working Example (5 lines)

import asyncio
from ai_prishtina_microsoft_graph_api import GraphAPIClient

async def main():
    async with GraphAPIClient(
        tenant_id="your-tenant-id",
        client_id="your-client-id",
        client_secret="your-client-secret"
    ) as client:
        users = await client.users.list_users()
        print(f"Found {len(users)} users")

asyncio.run(main())

Common Use Cases

await client.mail.send_message(
    subject="Hello from Graph API",
    body="<h1>Welcome!</h1><p>This is a test email.</p>",
    to_addresses=["recipient@example.com"],
    is_html=True
)

from datetime import datetime, timedelta

event = await client.calendar.create_event(
    subject="Team Meeting",
    start=datetime.now() + timedelta(hours=1),
    end=datetime.now() + timedelta(hours=2),
    attendees=["colleague@example.com"],
    location="Conference Room A"
)

from ai_prishtina_microsoft_graph_api.infrastructure.query import FilterBuilder, QueryParameters

# Build filter: department = 'Engineering' AND accountEnabled = true
filter_expr = (FilterBuilder()
    .equals("department", "Engineering")
    .and_()
    .equals("accountEnabled", True)
    .build())

params = QueryParameters(filter=filter_expr, top=50)
engineers = await client.users.list_users(query_params=params)

# Upload a file
with open("report.pdf", "rb") as f:
    file = await client.files.upload_file(
        folder_path="/Documents/Reports",
        filename="report.pdf",
        content=f.read()
    )
print(f"Uploaded: {file.web_url}")

🎯 Feature Overview

Core Capabilities

┌─────────────────────────────────────────────────────────────────────────────┐
│                        AI-PRISHTINA GRAPH API INGESTOR                       │
├─────────────────────────────────────────────────────────────────────────────┤
│                                                                             │
│  ╔═══════════════════════════════════════════════════════════════════════╗  │
│  ║  🔌 12 COMPLETE API INTEGRATIONS                                      ║  │
│  ║  ─────────────────────────────────────────────────────────────────────║  │
│  ║  📧 Mail        📅 Calendar     👥 Users       👔 Groups              ║  │
│  ║  💬 Teams       📁 Files        🌐 Sites       📋 Planner             ║  │
│  ║  ✅ To Do       📇 Contacts     📓 Notes       🏢 Directory           ║  │
│  ║  📊 Reports                                                           ║  │
│  ╚═══════════════════════════════════════════════════════════════════════╝  │
│                                                                             │
│  ╔═══════════════════════════════════════════════════════════════════════╗  │
│  ║  ⚡ ADVANCED FEATURES                                                 ║  │
│  ║  ─────────────────────────────────────────────────────────────────────║  │
│  ║  🔍 Query Parameters    │ OData $filter, $orderby, $select, $expand   ║  │
│  ║  📦 Batch Operations    │ Combine up to 20 API calls in one request  ║  │
│  ║  🔄 Delta Queries       │ Incremental sync for change tracking        ║  │
│  ╚═══════════════════════════════════════════════════════════════════════╝  │
│                                                                             │
│  ╔═══════════════════════════════════════════════════════════════════════╗  │
│  ║  🏢 ENTERPRISE FEATURES                                               ║  │
│  ║  ─────────────────────────────────────────────────────────────────────║  │
│  ║  🔐 OAuth2 Auth    │ Client credentials, auth code, device code       ║  │
│  ║  🔄 Retry Logic    │ Exponential backoff with configurable attempts   ║  │
│  ║  ⚡ Rate Limiting  │ Token bucket algorithm prevents throttling       ║  │
│  ║  🛡️ Circuit Breaker│ Fault tolerance, prevents cascading failures    ║  │
│  ║  💾 Redis Cache    │ Token & response caching with auto-refresh       ║  │
│  ║  📊 Prometheus     │ Full metrics for requests, errors, latency       ║  │
│  ║  📝 Structured Logs│ JSON logging with correlation IDs                ║  │
│  ╚═══════════════════════════════════════════════════════════════════════╝  │
│                                                                             │
└─────────────────────────────────────────────────────────────────────────────┘

🏗️ Clean Architecture

This library is built on Clean Architecture principles, ensuring separation of concerns, testability, and maintainability.

Architecture Diagram

graph TB
    subgraph "Presentation Layer"
        A[Pydantic Models]
        B[DTOs]
        C[API Contracts]
    end

    subgraph "Application Layer"
        D[Services]
        E[Use Cases]
        F[Workflows]
    end

    subgraph "Domain Layer"
        G[Entities]
        H[Value Objects]
        I[Repository Interfaces]
        J[Domain Exceptions]
    end

    subgraph "Infrastructure Layer"
        K[HTTP Client]
        L[Authentication]
        M[Cache - Redis/Memory]
        N[Rate Limiter]
        O[Circuit Breaker]
        P[Retry Handler]
        Q[Prometheus Metrics]
        R[Structured Logging]
    end

    A --> D
    B --> D
    D --> G
    D --> I
    E --> I
    I --> K
    K --> L
    K --> M
    K --> N
    K --> O
    K --> P
    K --> Q
    K --> R

    style A fill:#e1f5fe
    style B fill:#e1f5fe
    style C fill:#e1f5fe
    style D fill:#e8f5e9
    style E fill:#e8f5e9
    style F fill:#e8f5e9
    style G fill:#fff3e0
    style H fill:#fff3e0
    style I fill:#fff3e0
    style J fill:#fff3e0
    style K fill:#fce4ec
    style L fill:#fce4ec
    style M fill:#fce4ec
    style N fill:#fce4ec
    style O fill:#fce4ec
    style P fill:#fce4ec
    style Q fill:#fce4ec
    style R fill:#fce4ec

Layer Responsibilities

Directory Structure

ai_prishtina_microsoft_graph_api/
├── __init__.py              # Package exports
├── client.py                # Main GraphAPIClient
├── application/
│   └── services/            # Business logic services (12 APIs)
├── domain/
│   ├── entities/            # Domain entities (User, Event, etc.)
│   ├── exceptions/          # Domain exceptions
│   └── repositories/        # Repository interfaces
├── infrastructure/
│   ├── auth/                # OAuth2 token providers
│   ├── batch/               # Batch request handling
│   ├── cache/               # Redis & memory cache
│   ├── http/                # HTTP client with retry
│   ├── monitoring/          # Prometheus & logging
│   ├── pagination/          # Auto-pagination
│   ├── query/               # Query parameters & delta
│   ├── rate_limiting/       # Rate limiter & circuit breaker
│   └── repositories/        # Repository implementations
├── presentation/
│   └── models/              # Pydantic models
└── config/                  # Configuration management

📊 Complete API Coverage

All 12 Microsoft Graph API categories are fully implemented with CRUD operations, query parameters, batch operations, and (where applicable) delta queries.

🔥 Advanced Features

1. Query Parameters (OData Support)

Build complex queries with a fluent, type-safe API:

from ai_prishtina_microsoft_graph_api.infrastructure.query import (
    FilterBuilder, QueryParameters, OrderDirection
)

# Build complex filter expression
filter_expr = (FilterBuilder()
    .equals("department", "Engineering")
    .and_()
    .startswith("displayName", "A")
    .and_()
    .equals("accountEnabled", True)
    .build())
# Result: "department eq 'Engineering' and startswith(displayName, 'A') and accountEnabled eq true"

# Create query with all parameters
params = QueryParameters(
    filter=filter_expr,
    orderby=[("displayName", OrderDirection.ASC), ("createdDateTime", OrderDirection.DESC)],
    select=["id", "displayName", "mail", "department", "jobTitle"],
    expand=["manager", "directReports"],
    top=50,
    skip=0,
    count=True
)

# Execute query
users = await client.users.list_users(query_params=params)

2. Batch Operations

Combine up to 20 API calls in a single HTTP request:

from ai_prishtina_microsoft_graph_api.infrastructure.batch import BatchRequest

# Create batch with multiple operations
batch = BatchRequest()
id1 = batch.add_get("/users/user1@example.com")
id2 = batch.add_get("/users/user2@example.com")
id3 = batch.add_post("/users", {"displayName": "New User", "mailNickname": "newuser"})
id4 = batch.add_patch("/users/user3@example.com", {"department": "Engineering"})

# Execute all at once
response = await client.users.batch(batch)

# Process results
print(f"Success: {response.success_count}, Errors: {response.error_count}")
for item in response.responses:
    if item.is_success:
        print(f"✅ {item.id}: {item.status}")
    else:
        print(f"❌ {item.id}: {item.status} - {item.body}")

3. Delta Queries (Change Tracking)

Efficiently sync changes with incremental updates:

from ai_prishtina_microsoft_graph_api.infrastructure.query import DeltaState

# Initial sync - get all users
state = DeltaState(resource_type="users")
result = await client.users.delta()
state.update(result.delta_token, len(result.items))
print(f"Initial sync: {len(result.items)} users")

# Save delta_token to persistent storage...

# Later: Incremental sync - only changes since last sync
result = await client.users.delta(delta_token=state.delta_token)
print(f"Changes since last sync: {len(result.items)}")
for user in result.items:
    if user.is_deleted:
        print(f"  Deleted: {user.id}")
    else:
        print(f"  Modified: {user.display_name}")

🏢 Enterprise Features

Authentication Flow

sequenceDiagram
    participant App as Your Application
    participant Client as GraphAPIClient
    participant TokenProvider as Token Provider
    participant Cache as Redis/Memory Cache
    participant AzureAD as Azure AD
    participant GraphAPI as Microsoft Graph API

    App->>Client: Initialize with credentials
    Client->>TokenProvider: Create provider

    App->>Client: Make API request
    Client->>TokenProvider: Get token
    TokenProvider->>Cache: Check cached token

    alt Token cached and valid
        Cache-->>TokenProvider: Return cached token
    else Token missing or expired
        TokenProvider->>AzureAD: Request new token
        AzureAD-->>TokenProvider: Return token
        TokenProvider->>Cache: Store token
    end

    TokenProvider-->>Client: Return token
    Client->>GraphAPI: API request with token
    GraphAPI-->>Client: Response
    Client-->>App: Return result

Request Lifecycle with Resilience

flowchart TD
    A[API Request] --> B{Rate Limiter}
    B -->|Allowed| C{Circuit Breaker}
    B -->|Throttled| D[Wait & Retry]
    D --> B

    C -->|Closed| E[Make HTTP Request]
    C -->|Open| F[Fail Fast]
    C -->|Half-Open| G[Test Request]

    E --> H{Response?}
    G --> H

    H -->|Success| I[Return Response]
    H -->|Transient Error| J{Retry Handler}
    H -->|Permanent Error| K[Throw Exception]

    J -->|Retry Available| L[Exponential Backoff]
    J -->|Max Retries| K
    L --> E

    I --> M[Update Metrics]
    K --> M
    F --> M

    style A fill:#e3f2fd
    style I fill:#c8e6c9
    style K fill:#ffcdd2
    style F fill:#ffcdd2

Circuit Breaker States

from ai_prishtina_microsoft_graph_api import GraphAPIClient

async with GraphAPIClient(...) as client:
    # Access circuit breaker stats
    stats = client.container.circuit_breaker().get_stats()
    print(f"""
    Circuit State: {stats['state']}
    Failure Count: {stats['failure_count']}
    Failure Threshold: {stats['failure_threshold']}
    Recovery Timeout: {stats['recovery_timeout']}s
    Last Failure: {stats['last_failure_time']}
    """)

Rate Limiting

Token bucket algorithm prevents API throttling:

# Configuration (in GraphAPIClient or config file)
# Default: 10 requests/second with burst of 20

# Access rate limiter stats
rate_limiter = client.container.rate_limiter()
# Rate limiter automatically manages request flow

⚙️ Configuration

Option 1: Environment Variables

# Required: Azure AD credentials
export AZURE_TENANT_ID="your-tenant-id"
export AZURE_CLIENT_ID="your-client-id"
export AZURE_CLIENT_SECRET="your-client-secret"

# Optional: API configuration
export GRAPH_API_BASE_URL="https://graph.microsoft.com/v1.0"
export GRAPH_API_TIMEOUT="30"

# Optional: Redis caching
export REDIS_ENABLED="true"
export REDIS_HOST="localhost"
export REDIS_PORT="6379"
export REDIS_PASSWORD=""

# Optional: Rate limiting
export RATE_LIMIT_REQUESTS_PER_SECOND="10"
export RATE_LIMIT_BURST_SIZE="20"

# Optional: Circuit breaker
export CIRCUIT_BREAKER_FAILURE_THRESHOLD="5"
export CIRCUIT_BREAKER_RECOVERY_TIMEOUT="60"

Option 2: Configuration File (YAML)

# config.yaml
authentication:
  tenant_id: "your-tenant-id"
  client_id: "your-client-id"
  client_secret: "your-client-secret"

api:
  base_url: "https://graph.microsoft.com/v1.0"
  timeout_seconds: 30

redis:
  enabled: true
  host: "localhost"
  port: 6379
  db: 0
  password: null
  ttl_seconds: 3600

rate_limiting:
  enabled: true
  requests_per_second: 10.0
  burst_size: 20

circuit_breaker:
  enabled: true
  failure_threshold: 5
  recovery_timeout_seconds: 60

retry:
  enabled: true
  max_attempts: 5
  backoff_multiplier: 2.0
  max_wait_seconds: 60

monitoring:
  prometheus_enabled: true
  log_level: "INFO"
  log_format: "json"

# Load from config file
client = GraphAPIClient.from_config("config.yaml")

Option 3: Constructor Parameters

client = GraphAPIClient(
    # Authentication (required)
    tenant_id="your-tenant-id",
    client_id="your-client-id",
    client_secret="your-client-secret",

    # API configuration
    base_url="https://graph.microsoft.com/v1.0",

    # Redis caching
    use_redis=True,
    redis_host="localhost",
    redis_port=6379,
    redis_db=0,
    redis_password=None,
)

📊 Monitoring & Observability

Prometheus Metrics

from prometheus_client import start_http_server

# Start metrics server on port 8000
start_http_server(8000)

# Metrics available at http://localhost:8000/metrics

Structured Logging

from ai_prishtina_microsoft_graph_api.infrastructure.monitoring import configure_logging

# JSON logs for production (ELK, Splunk, etc.)
configure_logging(level="INFO", json_logs=True)

# Human-readable logs for development
configure_logging(level="DEBUG", json_logs=False)

Example JSON log output:

{
  "timestamp": "2024-01-15T10:30:45.123Z",
  "level": "info",
  "logger": "ai_prishtina_microsoft_graph_api.infrastructure.http",
  "message": "graph_api_request",
  "method": "GET",
  "url": "/users",
  "status_code": 200,
  "duration_ms": 145,
  "correlation_id": "abc-123-def-456"
}

🏆 Competitive Advantages

vs. msgraph-sdk-python (Microsoft Official)

vs. O365 / python-o365

Why Choose AI-Prishtina?

1. 🚀 Performance: Native async design handles thousands of concurrent requests
2. 🛡️ Reliability: Enterprise patterns (circuit breaker, retry, rate limiting) ensure stability
3. 📊 Observability: Full Prometheus metrics and structured logging for production
4. 🧩 Maintainability: Clean Architecture makes code testable and extensible
5. ⚡ Efficiency: Batch operations and delta queries minimize API calls
6. 🔒 Type Safety: Pydantic models catch errors at compile time

📖 Documentation

Quick Links

API Examples

Additional Resources

• Graph API Overview - Microsoft Graph fundamentals
• Custom Library Rationale - Why we built this
• Implementation Plan - Development roadmap
• Research Findings - Library comparison analysis

🧪 Testing

# Run all tests
pytest

# Run with coverage report
pytest --cov=ai_prishtina_microsoft_graph_api --cov-report=html

# Run specific test categories
pytest tests/unit/           # Unit tests only
pytest tests/integration/    # Integration tests only

# Run with verbose output
pytest -v

# Run specific test file
pytest tests/unit/test_users_repository.py

Test Coverage Goals

• Unit Tests: ≥90% coverage of business logic
• Integration Tests: All API endpoints tested
• Performance Tests: Batch and concurrent operation benchmarks

📄 License & Support

License

This is proprietary commercial software. See the LICENSE file for details.

Copyright © 2025 Alban Maxhuni, PhD. All rights reserved.

For licensing inquiries, contact: info@albanmaxhuni.com

Support

Acknowledgments

Built with industry-leading open source tools:

• httpx - Async HTTP client
• MSAL - Microsoft authentication
• Pydantic - Data validation
• Prometheus - Metrics and monitoring
• structlog - Structured logging
• tenacity - Retry logic