Python ile REST API Entegrasyonu Nasıl Yapılır?

Sistem yöneticilerinin işlerinin büyük bir kısmı artık sadece sunucu kurmak, servis başlatmak ya da log okumaktan ibaret değil. Modern altyapılarda onlarca farklı sistem birbiriyle konuşuyor, izleme araçları ticket sistemlerine bağlanıyor, deployment pipeline’ları bulut servislerini tetikliyor ve her şeyin merkezinde REST API’lar var. Python bu noktada sysadmin’in en güçlü silahlarından biri haline geliyor. Bu yazıda Python ile REST API entegrasyonunu gerçek dünya senaryolarıyla ele alacağız.

REST API Nedir, Neden Önemli?

REST (Representational State Transfer), HTTP protokolü üzerinden kaynaklara erişim sağlayan bir mimari yaklaşım. Sysadmin perspektifinden bakarsak REST API’lar şu anlama geliyor: Grafana’dan alert almak, PagerDuty’ye incident göndermek, JIRA’da ticket açmak, AWS EC2 instance başlatmak veya durdurmak, hepsini komut satırından veya bir script üzerinden otomatize etmek.

HTTP metodları şu şekilde çalışır:

GET: Kaynak okuma, veri çekme
POST: Yeni kaynak oluşturma
PUT: Mevcut kaynağı tamamen güncelleme
PATCH: Mevcut kaynağı kısmen güncelleme
DELETE: Kaynak silme

Python’da REST API entegrasyonu için temel kütüphane requests. Standart kütüphane ile gelmiyor ama pip üzerinden kurmak saniyeler alıyor.

pip install requests
pip install requests[security]  # SSL doğrulama için ekstra paketler

İlk Adım: Basit GET İsteği

En temel senaryodan başlayalım. Bir monitoring API’sından sunucu durumunu çekelim.

# Önce requests kütüphanesini test edelim
python3 -c "import requests; print(requests.__version__)"

import requests
import json

# Basit bir GET isteği
def get_server_status(base_url, server_id, api_token):
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Content-Type": "application/json",
        "Accept": "application/json"
    }
    
    endpoint = f"{base_url}/api/v1/servers/{server_id}/status"
    
    try:
        response = requests.get(endpoint, headers=headers, timeout=10)
        response.raise_for_status()  # 4xx/5xx hatalarını yakala
        
        data = response.json()
        return data
        
    except requests.exceptions.ConnectionError:
        print(f"[HATA] {base_url} adresine bağlanılamadı")
        return None
    except requests.exceptions.Timeout:
        print("[HATA] İstek zaman aşımına uğradı")
        return None
    except requests.exceptions.HTTPError as e:
        print(f"[HATA] HTTP hatası: {e.response.status_code}")
        return None

# Kullanım
status = get_server_status(
    "https://monitoring.sirket.local",
    "web-prod-01",
    "your-api-token-here"
)

if status:
    print(f"Sunucu durumu: {status.get('state', 'Bilinmiyor')}")
    print(f"CPU kullanımı: {status.get('cpu_percent', 'N/A')}%")
    print(f"RAM kullanımı: {status.get('memory_percent', 'N/A')}%")

Burada dikkat edilmesi gereken birkaç kritik nokta var. timeout parametresini her zaman belirt, belirtmezsen script sonsuza kadar bekleyebilir. raise_for_status() metodu HTTP hata kodlarını exception’a çeviriyor, bunu kullanmak hata yönetimini ciddi ölçüde kolaylaştırıyor.

Kimlik Doğrulama Yöntemleri

Gerçek dünyada API’ların büyük çoğunluğu kimlik doğrulama istiyor. En yaygın yöntemler şunlar:

Bearer Token ile Kimlik Doğrulama

import requests
import os

# Token'ı environment variable'dan al, asla koda gömme
API_TOKEN = os.environ.get("MONITORING_API_TOKEN")

if not API_TOKEN:
    raise ValueError("MONITORING_API_TOKEN environment variable tanımlanmamış!")

session = requests.Session()
session.headers.update({
    "Authorization": f"Bearer {API_TOKEN}",
    "Content-Type": "application/json"
})

# Session kullanımı - birden fazla istek için daha verimli
response = session.get("https://api.ornek.com/v1/hosts")

Basic Authentication

import requests
from requests.auth import HTTPBasicAuth
import os

username = os.environ.get("API_USERNAME")
password = os.environ.get("API_PASSWORD")

# Yöntem 1: HTTPBasicAuth kullanımı
response = requests.get(
    "https://jenkins.sirket.local/api/json",
    auth=HTTPBasicAuth(username, password),
    verify=True  # SSL sertifikasını doğrula
)

# Yöntem 2: Kısa yazım
response = requests.get(
    "https://jenkins.sirket.local/api/json",
    auth=(username, password)
)

# İç network'teyseniz ve self-signed sertifika varsa
response = requests.get(
    "https://jenkins.internal/api/json",
    auth=(username, password),
    verify="/etc/ssl/certs/internal-ca.pem"  # CA sertifikası yolu
)

API Key Kullanımı

Bazı API’lar token’ı header yerine query parameter olarak bekliyor:

import requests

def call_api_with_key(endpoint, api_key, params=None):
    if params is None:
        params = {}
    
    # API key'i query parametresi olarak ekle
    params["api_key"] = api_key
    
    response = requests.get(endpoint, params=params, timeout=15)
    return response.json()

# URL şu şekilde oluşacak: https://api.ornek.com/data?api_key=xxx&filter=active
data = call_api_with_key(
    "https://api.ornek.com/data",
    api_key="gizli-api-key",
    params={"filter": "active", "limit": 100}
)

POST ile Veri Gönderme: JIRA Ticket Oluşturma

Pratik bir senaryo: Monitoring sistemi alert ürettiğinde otomatik olarak JIRA ticket açmak. Bu tür otomasyon sysadmin’in gecenin 3’ünde telefona bakmak zorunda kalmasını engelliyor.

import requests
import json
import os
from datetime import datetime

def create_jira_ticket(summary, description, priority="Medium", project_key="OPS"):
    """
    JIRA REST API v3 kullanarak ticket oluşturur
    """
    jira_url = os.environ.get("JIRA_URL")
    jira_email = os.environ.get("JIRA_EMAIL")
    jira_token = os.environ.get("JIRA_API_TOKEN")
    
    endpoint = f"{jira_url}/rest/api/3/issue"
    
    # JIRA API payload formatı
    payload = {
        "fields": {
            "project": {
                "key": project_key
            },
            "summary": summary,
            "description": {
                "type": "doc",
                "version": 1,
                "content": [
                    {
                        "type": "paragraph",
                        "content": [
                            {
                                "type": "text",
                                "text": description
                            }
                        ]
                    }
                ]
            },
            "issuetype": {
                "name": "Incident"
            },
            "priority": {
                "name": priority
            },
            "labels": ["auto-created", "monitoring-alert"]
        }
    }
    
    headers = {
        "Accept": "application/json",
        "Content-Type": "application/json"
    }
    
    try:
        response = requests.post(
            endpoint,
            headers=headers,
            auth=(jira_email, jira_token),
            data=json.dumps(payload),
            timeout=30
        )
        response.raise_for_status()
        
        ticket_data = response.json()
        ticket_key = ticket_data.get("key")
        
        print(f"[OK] JIRA ticket oluşturuldu: {ticket_key}")
        print(f"[OK] URL: {jira_url}/browse/{ticket_key}")
        
        return ticket_key
        
    except requests.exceptions.HTTPError as e:
        print(f"[HATA] JIRA ticket oluşturulamadı: {e}")
        print(f"[HATA] Response: {e.response.text}")
        return None

# Monitoring alertinden gelen verilerle kullanım
alert_hostname = "db-prod-02"
alert_message = "CPU kullanımı %95 üzerinde, son 10 dakika"
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

ticket_key = create_jira_ticket(
    summary=f"[ALERT] {alert_hostname} - Yüksek CPU Kullanımı",
    description=f"Zaman: {timestamp}nSunucu: {alert_hostname}nDetay: {alert_message}",
    priority="High"
)

Sayfalama (Pagination) Yönetimi

Gerçek API entegrasyonlarında karşılaştığın en yaygın sorunlardan biri büyük veri setlerini çekmek. API’lar genellikle bir seferde 100 veya 1000 kayıt döndürür, daha fazlası için sayfalama kullanman gerekir.

import requests
import time

def get_all_hosts(base_url, api_token):
    """
    Tüm host'ları sayfalama yaparak çeker
    """
    all_hosts = []
    page = 1
    per_page = 100
    
    headers = {
        "Authorization": f"Bearer {api_token}",
        "Accept": "application/json"
    }
    
    while True:
        params = {
            "page": page,
            "per_page": per_page,
            "status": "active"
        }
        
        try:
            response = requests.get(
                f"{base_url}/api/v1/hosts",
                headers=headers,
                params=params,
                timeout=30
            )
            response.raise_for_status()
            
            data = response.json()
            hosts = data.get("hosts", [])
            
            if not hosts:
                print(f"[INFO] Toplam {len(all_hosts)} host çekildi")
                break
            
            all_hosts.extend(hosts)
            print(f"[INFO] Sayfa {page}: {len(hosts)} host çekildi")
            
            # Toplam sayfa sayısını kontrol et
            total_pages = data.get("total_pages", 1)
            if page >= total_pages:
                break
            
            page += 1
            
            # Rate limiting için kısa bekleme
            time.sleep(0.5)
            
        except requests.exceptions.RequestException as e:
            print(f"[HATA] Sayfa {page} çekilemedi: {e}")
            break
    
    return all_hosts

# Tüm host listesini çek ve işle
hosts = get_all_hosts("https://cmdb.sirket.local", "api-token-buraya")

# İşletim sistemine göre grupla
linux_hosts = [h for h in hosts if h.get("os_type") == "linux"]
windows_hosts = [h for h in hosts if h.get("os_type") == "windows"]

print(f"Linux sunucu sayısı: {len(linux_hosts)}")
print(f"Windows sunucu sayısı: {len(windows_hosts)}")

Rate Limiting ve Retry Mekanizması

Ticari API’ların büyük kısmı rate limiting uyguluyor. Çok fazla istek atarsan 429 (Too Many Requests) hatası alırsın. Bunun için retry mekanizması kurmak şart.

import requests
import time
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def create_session_with_retry():
    """
    Otomatik retry özellikli requests session oluşturur
    """
    session = requests.Session()
    
    # Retry stratejisi
    retry_strategy = Retry(
        total=5,                  # Maksimum retry sayısı
        backoff_factor=1,         # Her retry'da bekleme süresi: 1s, 2s, 4s, 8s...
        status_forcelist=[        # Bu HTTP kodlarında retry yap
            429,  # Too Many Requests
            500,  # Internal Server Error
            502,  # Bad Gateway
            503,  # Service Unavailable
            504   # Gateway Timeout
        ],
        allowed_methods=["GET", "POST", "PUT", "DELETE"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

def api_request_with_rate_limit(session, url, headers, method="GET", payload=None):
    """
    Rate limit header'ını kontrol ederek istek atar
    """
    try:
        if method == "GET":
            response = session.get(url, headers=headers, timeout=30)
        elif method == "POST":
            response = session.post(url, headers=headers, json=payload, timeout=30)
        
        # Rate limit bilgisini logla
        remaining = response.headers.get("X-RateLimit-Remaining", "N/A")
        reset_time = response.headers.get("X-RateLimit-Reset", "N/A")
        
        if remaining != "N/A" and int(remaining) < 10:
            print(f"[UYARI] Rate limit yaklaşıyor! Kalan istek: {remaining}")
        
        # Manuel rate limit yönetimi
        if response.status_code == 429:
            retry_after = int(response.headers.get("Retry-After", 60))
            print(f"[UYARI] Rate limit aşıldı. {retry_after} saniye bekleniyor...")
            time.sleep(retry_after)
            # Tekrar dene
            return api_request_with_rate_limit(session, url, headers, method, payload)
        
        response.raise_for_status()
        return response.json()
        
    except requests.exceptions.RequestException as e:
        print(f"[HATA] İstek başarısız: {e}")
        return None

# Kullanım
session = create_session_with_retry()
headers = {"Authorization": "Bearer token-buraya"}

data = api_request_with_rate_limit(
    session,
    "https://api.github.com/repos/sirket/proje/issues",
    headers
)

Slack Webhook ile Alert Gönderme

Monitoring alert’lerini Slack’e göndermek son derece yaygın bir kullanım senaryosu. Bu senaryo sysadmin’lerin %80’inin hayatında bir noktada karşılaştığı bir durum.

import requests
import json
import os

def send_slack_alert(message, severity="info", channel=None, hostname=None):
    """
    Slack Incoming Webhook ile alert gönderir
    """
    webhook_url = os.environ.get("SLACK_WEBHOOK_URL")
    
    if not webhook_url:
        print("[HATA] SLACK_WEBHOOK_URL environment variable tanımlanmamış!")
        return False
    
    # Severity'ye göre renk belirle
    color_map = {
        "critical": "#FF0000",  # Kırmızı
        "warning": "#FFA500",   # Turuncu
        "info": "#36A64F",      # Yeşil
        "resolved": "#439FE0"   # Mavi
    }
    
    color = color_map.get(severity, "#808080")
    
    # Slack mesaj formatı
    payload = {
        "attachments": [
            {
                "color": color,
                "title": f"[{severity.upper()}] Sistem Alarmı",
                "text": message,
                "fields": [
                    {
                        "title": "Sunucu",
                        "value": hostname or "Bilinmiyor",
                        "short": True
                    },
                    {
                        "title": "Ortam",
                        "value": os.environ.get("ENVIRONMENT", "production"),
                        "short": True
                    }
                ],
                "footer": "Monitoring System",
                "ts": int(__import__('time').time())
            }
        ]
    }
    
    if channel:
        payload["channel"] = channel
    
    try:
        response = requests.post(
            webhook_url,
            data=json.dumps(payload),
            headers={"Content-Type": "application/json"},
            timeout=15
        )
        response.raise_for_status()
        
        if response.text == "ok":
            print(f"[OK] Slack mesajı gönderildi: {severity.upper()}")
            return True
            
    except requests.exceptions.RequestException as e:
        print(f"[HATA] Slack mesajı gönderilemedi: {e}")
        return False

# Kullanım örnekleri
send_slack_alert(
    message="Disk kullanımı %90 üzerine çıktı! /var/log dizini temizlenmeli.",
    severity="warning",
    hostname="web-prod-03"
)

send_slack_alert(
    message="Database bağlantısı koptu! Otomatik failover başlatıldı.",
    severity="critical",
    hostname="db-prod-01"
)

send_slack_alert(
    message="Disk kullanımı normale döndü (%75). Alarm kapatıldı.",
    severity="resolved",
    hostname="web-prod-03"
)

Response Cache ve Performans Optimizasyonu

Sık sık aynı endpoint’i sorgulayan script’lerde gereksiz API çağrısı yapmaktan kaçınmak için basit bir cache mekanizması kurabilirsin.

import requests
import time
from functools import wraps

# Basit in-memory cache
_cache = {}

def cached_api_call(ttl_seconds=300):
    """
    API çağrılarını önbelleğe alan decorator
    ttl_seconds: Cache geçerlilik süresi (saniye)
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # Cache key oluştur
            cache_key = f"{func.__name__}:{str(args)}:{str(kwargs)}"
            
            # Cache'de var mı ve hala geçerli mi?
            if cache_key in _cache:
                cached_time, cached_data = _cache[cache_key]
                if time.time() - cached_time < ttl_seconds:
                    print(f"[CACHE] {func.__name__} cache'den döndürülüyor")
                    return cached_data
            
            # Cache'de yoksa API'yı çağır
            result = func(*args, **kwargs)
            
            if result is not None:
                _cache[cache_key] = (time.time(), result)
            
            return result
        return wrapper
    return decorator

@cached_api_call(ttl_seconds=600)  # 10 dakika cache
def get_infrastructure_list(base_url, api_token):
    """Altyapı listesini çeker - sık değişmediği için cache'liyoruz"""
    headers = {"Authorization": f"Bearer {api_token}"}
    
    response = requests.get(
        f"{base_url}/api/v1/infrastructure",
        headers=headers,
        timeout=30
    )
    response.raise_for_status()
    return response.json()

@cached_api_call(ttl_seconds=60)  # 1 dakika cache
def get_current_alerts(base_url, api_token):
    """Aktif alertleri çeker - daha sık cache yenileme gerekli"""
    headers = {"Authorization": f"Bearer {api_token}"}
    
    response = requests.get(
        f"{base_url}/api/v1/alerts/active",
        headers=headers,
        timeout=15
    )
    response.raise_for_status()
    return response.json()

# İlk çağrı API'ya gider
infra = get_infrastructure_list("https://cmdb.local", "token")
# İkinci çağrı cache'den gelir
infra = get_infrastructure_list("https://cmdb.local", "token")

Gerçek Dünya Senaryosu: AWS EC2 Yönetimi

AWS API’sını doğrudan requests ile kullanmak yerine genellikle boto3 kütüphanesi tercih edilse de bazen farklı bulut sağlayıcılarının API’larıyla çalışmak gerekiyor. Burada DigitalOcean API örneği ile bir Droplet yönetim scripti yazalım.

import requests
import os
import sys

class DigitalOceanAPI:
    """DigitalOcean API v2 wrapper sınıfı"""
    
    BASE_URL = "https://api.digitalocean.com/v2"
    
    def __init__(self, token=None):
        self.token = token or os.environ.get("DO_API_TOKEN")
        if not self.token:
            raise ValueError("DigitalOcean API token bulunamadı!")
        
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.token}",
            "Content-Type": "application/json"
        })
    
    def list_droplets(self, tag=None):
        """Tüm Droplet'ları listeler, opsiyonel tag filtresi"""
        params = {"per_page": 200}
        if tag:
            params["tag_name"] = tag
        
        response = self.session.get(
            f"{self.BASE_URL}/droplets",
            params=params,
            timeout=30
        )
        response.raise_for_status()
        return response.json().get("droplets", [])
    
    def get_droplet(self, droplet_id):
        """Belirli bir Droplet'ın detaylarını getirir"""
        response = self.session.get(
            f"{self.BASE_URL}/droplets/{droplet_id}",
            timeout=30
        )
        response.raise_for_status()
        return response.json().get("droplet")
    
    def power_action(self, droplet_id, action):
        """Droplet güç işlemi: power_on, power_off, reboot"""
        valid_actions = ["power_on", "power_off", "reboot", "shutdown"]
        if action not in valid_actions:
            raise ValueError(f"Geçersiz aksiyon. Kullanılabilir: {valid_actions}")
        
        payload = {"type": action}
        
        response = self.session.post(
            f"{self.BASE_URL}/droplets/{droplet_id}/actions",
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        
        action_data = response.json().get("action", {})
        print(f"[OK] {action} aksiyonu başlatıldı. Action ID: {action_data.get('id')}")
        return action_data
    
    def create_snapshot(self, droplet_id, snapshot_name):
        """Droplet snapshot'ı oluşturur"""
        payload = {
            "type": "snapshot",
            "name": snapshot_name
        }
        
        response = self.session.post(
            f"{self.BASE_URL}/droplets/{droplet_id}/actions",
            json=payload,
            timeout=30
        )
        response.raise_for_status()
        return response.json().get("action", {})


# Kullanım örneği
if __name__ == "__main__":
    do = DigitalOceanAPI()
    
    # Production tag'li tüm sunucuları listele
    prod_droplets = do.list_droplets(tag="production")
    
    print(f"Production sunucu sayısı: {len(prod_droplets)}")
    
    for droplet in prod_droplets:
        name = droplet.get("name")
        status = droplet.get("status")
        memory = droplet.get("memory", 0) // 1024  # MB'yi GB'ye çevir
        
        print(f"  - {name}: {status} ({memory}GB RAM)")
        
        # Kapalı olan production sunucuları uyar
        if status == "off":
            print(f"    [UYARI] {name} kapalı durumda!")

Sonuç

Python ile REST API entegrasyonu, modern sysadmin’in temel becerilerinden biri haline geldi. Yukarıda ele aldığımız konular bir araya geldiğinde ciddi otomasyon altyapıları kurabilirsin.

Önemli hatırlatmalar:

API token’larını asla koda gömme, environment variable veya secrets manager kullan
Her zaman timeout belirle, belirtmezsen script’in askıda kalma riski var
raise_for_status() kullan, HTTP hataları sessizce geçmesin
Rate limiting’e saygı göster, API sağlayıcılar IP banlamaktan çekinmiyor
Retry mekanizması kur, ağ sorunları geçici olabilir
Loglama ihmal etme, 3 ay sonra script’e baktığında ne yaptığını anlayabilesin

requests kütüphanesi başlangıç için mükemmel ama daha büyük projeler için httpx (async desteği var) veya GraphQL entegrasyonları için özel kütüphanelere bakmak da değer. Temel kavramları sindirdikten sonra hangi API’yla çalışırsan çalış, mantık aynı kalıyor: doğru header, doğru payload, doğru hata yönetimi.

Python ile REST API Entegrasyonu: Kapsamlı Bir Rehber