REST API ile Sistem Yönetimi: Python requests Modülü

Sistem yönetimi işlerinde bir noktadan sonra elle yapılan her şeyin bir maliyeti olduğunu fark ediyorsunuz. Sunucu durumu kontrol etmek, servis yönetimi, kullanıcı işlemleri… Bunların hepsini web arayüzünden tıklayarak yapmak yerine, REST API’leri Python ile otomatize etmek hem zaman kazandırıyor hem de insan hatasını ortadan kaldırıyor. requests modülü bu noktada sysadmin’in en güvenilir silahlarından biri haline geliyor.

requests Modülü Nedir ve Neden Kullanmalısınız

Python’un standart kütüphanesinde HTTP işlemleri için urllib var, ancak bunu kullananlar bilir ki oldukça karmaşık ve verbose bir yapısı var. requests modülü bu sorunu “HTTP for Humans” sloganıyla çözmek için geliştirildi. Syntax’ı temiz, hata yönetimi mantıklı, session yönetimi güçlü.

Sysadmin perspektifinden bakıldığında requests modülünün öne çıkan özellikleri şunlar:

• Session yönetimi: Aynı sunucuya birden fazla istek atarken cookie ve header’ları otomatik taşır
• SSL sertifika doğrulaması: Kurumsal ortamlarda özel CA’larla çalışabilme
• Timeout desteği: Askıda kalan isteklerin önüne geçme
• Retry mekanizması: Geçici ağ sorunlarını otomatik aşma
• JSON desteği: API yanıtlarını direkt parse etme

Kurulum için:

pip install requests
# veya sistem genelinde
pip3 install requests

# Sanal ortam tercih ederseniz
python3 -m venv sysadmin-env
source sysadmin-env/bin/activate
pip install requests

Temel HTTP İstekleri

Gerçek dünyada karşılaşacağınız senaryolara geçmeden önce temel yapıyı netleştirelim.

import requests
import json

# GET isteği - örnek: sunucu health check
response = requests.get(
    'https://api.sunucumonitoring.local/health',
    headers={'Authorization': 'Bearer TOKEN_BURAYA'},
    timeout=10,
    verify='/etc/ssl/certs/internal-ca.pem'
)

# Yanıt kodunu kontrol et
if response.status_code == 200:
    data = response.json()
    print(f"Sunucu durumu: {data['status']}")
elif response.status_code == 401:
    print("Kimlik doğrulama hatası, token'ı kontrol edin")
elif response.status_code == 503:
    print("Servis kullanılamıyor!")

# POST isteği - örnek: yeni kullanıcı oluşturma
payload = {
    'username': 'ahmet.yilmaz',
    'email': '[email protected]',
    'role': 'readonly'
}

response = requests.post(
    'https://api.sunucumonitoring.local/users',
    json=payload,
    headers={'Authorization': 'Bearer TOKEN_BURAYA'},
    timeout=30
)

print(f"HTTP Status: {response.status_code}")
print(f"Yanıt: {response.json()}")

Burada dikkat edilmesi gereken birkaç nokta var. timeout parametresini her zaman kullanın. Üretim ortamında timeout vermeden request atan script’ler, servis yanıt vermediğinde sonsuza kadar bekler ve bu durum kascade failure’lara yol açabilir. verify parametresiyle kendi CA’nızın sertifikasını belirtebilirsiniz; verify=False kullanmaktan kaçının, bu güvenlik açığı yaratır.

Session Kullanımı: Aynı Sunucuya Çoklu İstekler

Birden fazla endpoint’e istek atacaksanız Session nesnesi hem performans hem de kod temizliği açısından çok daha iyi bir seçenek.

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

def api_session_olustur(base_url, token, ca_cert_path=None):
    """
    Retry mekanizması ve authentication ile
    hazır bir session nesnesi döndürür.
    """
    session = requests.Session()
    
    # Authentication header'ı tüm isteklere ekle
    session.headers.update({
        'Authorization': f'Bearer {token}',
        'Content-Type': 'application/json',
        'User-Agent': 'SysadminBot/1.0'
    })
    
    # SSL sertifika ayarları
    if ca_cert_path:
        session.verify = ca_cert_path
    
    # Retry stratejisi: 3 deneme, 429/500/502/503/504 için
    retry_strategy = Retry(
        total=3,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["HEAD", "GET", "PUT", "DELETE", "OPTIONS", "TRACE"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    session.mount("http://", adapter)
    
    return session

# Kullanım örneği
SESSION = api_session_olustur(
    base_url='https://monitoring.sirket.local',
    token='eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9...',
    ca_cert_path='/etc/ssl/certs/sirket-ca-bundle.pem'
)

# Artık her istekte ayrıca header belirtmeye gerek yok
r1 = SESSION.get('https://monitoring.sirket.local/api/servers')
r2 = SESSION.get('https://monitoring.sirket.local/api/alerts')
r3 = SESSION.get('https://monitoring.sirket.local/api/metrics')

Retry stratejisini dikkatli kurun. POST gibi idempotent olmayan metodları retry listesine eklemeyin, çünkü ağ hatası aldıktan sonra aslında istek sunucuya ulaşmış olabilir ve tekrar gönderince duplicate kayıt oluşturabilirsiniz.

Gerçek Dünya Senaryosu 1: VMware vCenter REST API

Sanallaştırma ortamlarında VM yönetimi için vCenter REST API oldukça yaygın kullanılıyor. Örnek bir senaryo: Gece bakım penceresi için belirli VM’leri kapatıp sabah açmak.

import requests
import sys
import logging
from datetime import datetime

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(levelname)s - %(message)s',
    handlers=[
        logging.FileHandler('/var/log/vm-maintenance.log'),
        logging.StreamHandler()
    ]
)
logger = logging.getLogger(__name__)

VCENTER_URL = 'https://vcenter.sirket.local'
BAKM_VM_LISTESI = ['prod-web-01', 'prod-web-02', 'prod-app-01']

def vcenter_token_al(vcenter_url, kullanici, sifre):
    """vCenter API token alır."""
    response = requests.post(
        f'{vcenter_url}/api/session',
        auth=(kullanici, sifre),
        verify='/etc/ssl/certs/vcenter-ca.pem',
        timeout=30
    )
    
    if response.status_code != 201:
        raise Exception(f"Token alınamadı: {response.status_code} - {response.text}")
    
    token = response.json()
    logger.info("vCenter token başarıyla alındı")
    return token

def vm_listesi_al(session, vcenter_url):
    """Tüm VM listesini çeker."""
    response = session.get(
        f'{vcenter_url}/api/vcenter/vm',
        timeout=60
    )
    response.raise_for_status()
    return response.json()

def vm_kapat(session, vcenter_url, vm_id, vm_adi):
    """Belirtilen VM'i düzgünce kapatır."""
    logger.info(f"VM kapatılıyor: {vm_adi} ({vm_id})")
    
    # Önce guest shutdown dene (OS'u düzgün kapat)
    response = session.post(
        f'{vcenter_url}/api/vcenter/vm/{vm_id}/guest/power',
        json={'action': 'shutdown'},
        timeout=30
    )
    
    if response.status_code == 200:
        logger.info(f"{vm_adi} için guest shutdown başlatıldı")
        return True
    else:
        # Guest tools yoksa power off yap
        logger.warning(f"{vm_adi} guest shutdown başarısız, hard power off yapılıyor")
        response = session.post(
            f'{vcenter_url}/api/vcenter/vm/{vm_id}/power',
            json={'action': 'stop'},
            timeout=30
        )
        response.raise_for_status()
        return True

def bakm_penceresi_baslat(kullanici, sifre):
    """Bakım penceresi başlatır, listedeki VM'leri kapatır."""
    logger.info(f"Bakım penceresi başlatılıyor - {datetime.now()}")
    
    try:
        token = vcenter_token_al(VCENTER_URL, kullanici, sifre)
        
        session = requests.Session()
        session.headers.update({
            'vmware-api-session-id': token,
            'Content-Type': 'application/json'
        })
        session.verify = '/etc/ssl/certs/vcenter-ca.pem'
        
        tum_vmler = vm_listesi_al(session, VCENTER_URL)
        
        kapatilan = []
        hata_alan = []
        
        for vm in tum_vmler:
            if vm['name'] in BAKM_VM_LISTESI:
                try:
                    vm_kapat(session, VCENTER_URL, vm['vm'], vm['name'])
                    kapatilan.append(vm['name'])
                except Exception as e:
                    logger.error(f"{vm['name']} kapatılırken hata: {e}")
                    hata_alan.append(vm['name'])
        
        logger.info(f"Kapatılan VM'ler: {kapatilan}")
        if hata_alan:
            logger.error(f"Hata alan VM'ler: {hata_alan}")
            sys.exit(1)
            
    except Exception as e:
        logger.critical(f"Bakım penceresi başlatılamadı: {e}")
        sys.exit(1)

if __name__ == '__main__':
    bakm_penceresi_baslat('[email protected]', 'Sifre123!')

Gerçek Dünya Senaryosu 2: Alertmanager Silence API

Prometheus stack kullanıyorsanız, bakım sırasında gereksiz alarm gürültüsünü önlemek için Alertmanager’a silence eklemek hayat kurtarıyor. Bu işlemi otomatize etmek çok kullanışlı.

import requests
import json
from datetime import datetime, timedelta, timezone

ALERTMANAGER_URL = 'http://alertmanager.monitoring.svc:9093'

def silence_ekle(sunucu_adi, sure_dakika=120, yorum='Planlı bakım'):
    """
    Belirtilen sunucu için Alertmanager'a silence ekler.
    sure_dakika: Kaç dakika susturulacak
    """
    
    bitis_zamani = datetime.now(timezone.utc) + timedelta(minutes=sure_dakika)
    
    silence_payload = {
        "matchers": [
            {
                "name": "instance",
                "value": sunucu_adi,
                "isRegex": False
            }
        ],
        "startsAt": datetime.now(timezone.utc).isoformat(),
        "endsAt": bitis_zamani.isoformat(),
        "createdBy": "sysadmin-bot",
        "comment": yorum
    }
    
    response = requests.post(
        f'{ALERTMANAGER_URL}/api/v2/silences',
        json=silence_payload,
        timeout=15
    )
    
    if response.status_code == 200:
        silence_id = response.json()['silenceID']
        print(f"Silence eklendi: {silence_id}")
        print(f"Bitiş: {bitis_zamani.strftime('%Y-%m-%d %H:%M:%S UTC')}")
        return silence_id
    else:
        raise Exception(f"Silence eklenemedi: {response.status_code} - {response.text}")

def aktif_silence_listele():
    """Aktif tüm silence'ları listeler."""
    response = requests.get(
        f'{ALERTMANAGER_URL}/api/v2/silences',
        params={'filter': 'active'},
        timeout=15
    )
    response.raise_for_status()
    
    silences = response.json()
    aktif = [s for s in silences if s['status']['state'] == 'active']
    
    print(f"nAktif Silence Sayısı: {len(aktif)}")
    for s in aktif:
        print(f"  ID: {s['id'][:8]}...")
        print(f"  Yorum: {s['comment']}")
        print(f"  Bitiş: {s['endsAt']}")
        print(f"  Matchers: {s['matchers']}")
        print()
    
    return aktif

def silence_sil(silence_id):
    """Silence'ı erken sonlandırır."""
    response = requests.delete(
        f'{ALERTMANAGER_URL}/api/v2/silence/{silence_id}',
        timeout=15
    )
    
    if response.status_code == 200:
        print(f"Silence {silence_id} silindi")
    else:
        raise Exception(f"Silence silinemedi: {response.status_code}")

# Kullanım
silence_id = silence_ekle(
    sunucu_adi='prod-db-01.sirket.local',
    sure_dakika=240,
    yorum='Kernel update - 2024-01-15 gece bakımı'
)

Hata Yönetimi ve Exception Handling

Üretim ortamında çalışan script’lerde hata yönetimi kritik. requests modülünün fırlatabileceği exception’ları düzgün yakalamak gerekiyor.

import requests
from requests.exceptions import (
    ConnectionError,
    Timeout,
    TooManyRedirects,
    HTTPError,
    RequestException
)
import time
import logging

logger = logging.getLogger(__name__)

def guvenli_api_istegi(url, method='GET', **kwargs):
    """
    Kapsamlı hata yönetimiyle API isteği yapar.
    Geçici hatalarda retry uygular.
    """
    max_deneme = 3
    bekleme_suresi = 2
    
    for deneme in range(1, max_deneme + 1):
        try:
            response = requests.request(
                method=method,
                url=url,
                timeout=kwargs.pop('timeout', 30),
                **kwargs
            )
            
            # 4xx ve 5xx için exception fırlat
            response.raise_for_status()
            
            return response
            
        except Timeout:
            logger.warning(f"Timeout ({deneme}/{max_deneme}): {url}")
            if deneme == max_deneme:
                raise
            time.sleep(bekleme_suresi * deneme)
            
        except ConnectionError as e:
            logger.error(f"Bağlantı hatası ({deneme}/{max_deneme}): {e}")
            if deneme == max_deneme:
                raise
            time.sleep(bekleme_suresi * deneme)
            
        except HTTPError as e:
            status_code = e.response.status_code
            
            # 4xx hatalar retry'a gerek yok (client hatası)
            if 400 <= status_code < 500:
                logger.error(f"Client hatası {status_code}: {url}")
                logger.error(f"Yanıt: {e.response.text}")
                raise
            
            # 5xx hatalar için retry yap
            logger.warning(f"Server hatası {status_code} ({deneme}/{max_deneme}): {url}")
            if deneme == max_deneme:
                raise
            time.sleep(bekleme_suresi * deneme)
            
        except TooManyRedirects:
            logger.error(f"Çok fazla yönlendirme: {url}")
            raise
            
        except RequestException as e:
            logger.error(f"Beklenmedik hata: {e}")
            raise

# Kullanım
try:
    response = guvenli_api_istegi(
        'https://api.sirket.local/servers',
        method='GET',
        headers={'Authorization': 'Bearer TOKEN'},
        timeout=20
    )
    print(f"Başarılı: {response.json()}")
    
except Timeout:
    print("API yanıt vermedi, ops ekibini bilgilendir")
    
except ConnectionError:
    print("API sunucusuna ulaşılamıyor, network kontrolü yap")
    
except HTTPError as e:
    print(f"HTTP hatası: {e}")

Gerçek Dünya Senaryosu 3: GitLab/GitHub Üzerinden Config Dağıtımı

Infrastructure as Code çalışıyorsanız, konfigürasyon dosyalarını Git üzerinden çekip sunuculara dağıtmak yaygın bir pattern.

import requests
import os
import base64
import hashlib
from pathlib import Path

GITLAB_URL = 'https://gitlab.sirket.local'
GITLAB_TOKEN = os.environ.get('GITLAB_TOKEN')
PROJE_ID = '42'
DAL = 'main'

def gitlab_dosya_cek(dosya_yolu, proje_id=PROJE_ID, dal=DAL):
    """
    GitLab API üzerinden dosya içeriğini çeker.
    dosya_yolu: Repo içindeki yol, örn: 'configs/nginx/nginx.conf'
    """
    
    encoded_path = requests.utils.quote(dosya_yolu, safe='')
    
    response = requests.get(
        f'{GITLAB_URL}/api/v4/projects/{proje_id}/repository/files/{encoded_path}',
        headers={
            'PRIVATE-TOKEN': GITLAB_TOKEN,
            'Accept': 'application/json'
        },
        params={'ref': dal},
        verify='/etc/ssl/certs/gitlab-ca.pem',
        timeout=30
    )
    
    if response.status_code == 404:
        raise FileNotFoundError(f"Dosya bulunamadı: {dosya_yolu}")
    
    response.raise_for_status()
    
    data = response.json()
    icerik = base64.b64decode(data['content']).decode('utf-8')
    return icerik, data['content_sha256'], data['last_commit_id']

def config_dagit(dosya_yolu, hedef_sunucular, yerel_dizin='/etc/myapp'):
    """
    GitLab'dan config alıp hedef dizine yazar,
    değişiklik yoksa dokunmaz.
    """
    
    print(f"Config çekiliyor: {dosya_yolu}")
    icerik, git_sha256, commit_id = gitlab_dosya_cek(dosya_yolu)
    
    hedef_dosya = Path(yerel_dizin) / Path(dosya_yolu).name
    
    # Mevcut dosyayla karşılaştır
    if hedef_dosya.exists():
    
        with open(hedef_dosya, 'rb') as f:
            mevcut_hash = hashlib.sha256(f.read()).hexdigest()
        
        if mevcut_hash == git_sha256:
            print(f"Değişiklik yok, atlanıyor: {hedef_dosya}")
            return False
    
    # Yedek al
    if hedef_dosya.exists():
        yedek_yolu = f"{hedef_dosya}.bak"
        hedef_dosya.rename(yedek_yolu)
        print(f"Yedek alındı: {yedek_yolu}")
    
    # Yeni config'i yaz
    Path(yerel_dizin).mkdir(parents=True, exist_ok=True)
    with open(hedef_dosya, 'w') as f:
        f.write(icerik)
    
    print(f"Config güncellendi: {hedef_dosya}")
    print(f"Commit: {commit_id[:8]}")
    return True

# Dağıtılacak config listesi
KONFIGURASYON_LISTESI = [
    'configs/nginx/nginx.conf',
    'configs/nginx/ssl.conf',
    'configs/app/application.properties'
]

guncellenen = []
for config in KONFIGURASYON_LISTESI:
    try:
        degisti = config_dagit(config, yerel_dizin='/etc/nginx')
        if degisti:
            guncellenen.append(config)
    except Exception as e:
        print(f"HATA - {config}: {e}")

if guncellenen:
    print(f"n{len(guncellenen)} config güncellendi, servis yeniden başlatılabilir")
else:
    print("nHiçbir config değişmedi")

Environment Variable ile Güvenli Credential Yönetimi

Script’lere token ve şifre gömmek büyük güvenlik riski. Bunları environment variable veya secret manager’dan okumak standart pratik olmalı.

import os
import requests
from functools import lru_cache

@lru_cache(maxsize=1)
def config_yukle():
    """
    Ortam değişkenlerinden konfigürasyonu yükler.
    Eksik değer varsa açıklayıcı hata verir.
    """
    zorunlu_degiskenler = [
        'API_BASE_URL',
        'API_TOKEN',
        'SSL_CA_CERT'
    ]
    
    eksik = []
    config = {}
    
    for degisken in zorunlu_degiskenler:
        deger = os.environ.get(degisken)
        if not deger:
            eksik.append(degisken)
        else:
            config[degisken] = deger
    
    if eksik:
        raise EnvironmentError(
            f"Zorunlu environment variable eksik: {', '.join(eksik)}n"
            f"Örnek: export API_TOKEN='your-token-here'"
        )
    
    # İsteğe bağlı değişkenler varsayılanlarıyla
    config['TIMEOUT'] = int(os.environ.get('API_TIMEOUT', '30'))
    config['MAX_RETRY'] = int(os.environ.get('API_MAX_RETRY', '3'))
    
    return config

def api_baglanti_kur():
    """Konfigürasyondan session oluşturur."""
    config = config_yukle()
    
    session = requests.Session()
    session.headers.update({
        'Authorization': f"Bearer {config['API_TOKEN']}",
        'Content-Type': 'application/json'
    })
    session.verify = config['SSL_CA_CERT']
    
    return session, config

# .env dosyasından yüklemek isteyenler için (python-dotenv ile)
# from dotenv import load_dotenv
# load_dotenv('/etc/sysadmin-scripts/.env')

Systemd service olarak çalışan script’ler için environment variable’ları /etc/sysadmin-scripts/.env gibi bir dosyada tutup izinleri chmod 600 yapabilirsiniz. Systemd unit dosyasında EnvironmentFile=/etc/sysadmin-scripts/.env ile bu dosyayı yükleyebilirsiniz.

Sonuç

requests modülü, REST API’ler üzerinden sistem yönetimini profesyonel düzeye taşımanın en pratik yolu. Temel kullanımı öğrenmek bir saatten az sürüyor ama öğrendikten sonra her gün onlarca manuel işlemi otomatize edebiliyorsunuz.

Özetlemek gerekirse dikkat etmeniz gereken kritik noktalar:

• Her zaman timeout kullanın, aksi halde script’ler ağ sorunlarında sonsuza kadar bekler
• Session nesnesi kullanın aynı sunucuya birden fazla istek atacaksanız, hem performans hem temizlik açısından faydalı
• Credential’ları koda gömmeyın, environment variable veya secret manager kullanın
• raise_for_status() veya explicit status kontrol yapın, sessiz hatalar üretimde çok zarar verir
• Retry stratejisini dikkatli kurun, idempotent olmayan endpoint’ler için otomatik retry tehlikeli
• SSL doğrulamayı devre dışı bırakmayın, kurumsal CA sertifikanızı verify parametresine verin
• Log yazın, otomasyon script’lerinde ne olduğunu takip edebilmek sorun gidermede hayat kurtarıyor

Bu temelleri oturdurduktan sonra Prometheus API, Kubernetes API, Vault, Terraform Cloud, Ansible Tower gibi modern altyapı araçlarının hepsini Python script’lerinizden rahatça yönetebilirsiniz. Önemli olan her yeni entegrasyonda aynı pattern’i uygulamak: session oluştur, hata yönet, log yaz, credential güvenliğini sağla.