Sayfalama ile API Verilerini Toplu Çekme: Pagination Rehberi
Üretim ortamında binlerce, hatta milyonlarca kayıt döndüren bir API ile çalışmak zorunda kaldıysanız, muhtemelen şu sahneyle tanışıklığınız vardır: tek bir istek atarsınız, sunucu birkaç saniye düşünür, sonra ya timeout alırsınız ya da elinize devasa bir JSON yığını geçer ve uygulamanız bunu işlemeye çalışırken bellek çöpe gider. İşte tam bu noktada pagination, yani sayfalama devreye girer. API sayfalama, büyük veri setlerini yönetilebilir parçalara bölerek hem istemci hem de sunucu tarafında kaynak tüketimini dengeler. Bu yazıda pagination’ın ne olduğunu, farklı türlerini, gerçek dünya senaryolarında nasıl kullanacağınızı ve üretim kalitesinde Python/Bash scriptleri ile nasıl otomatize edeceğinizi ele alacağız.
Pagination Neden Gerekli?
Bir veritabanında 500.000 kullanıcı kaydı olan bir sistemden düşünün. İstemci her seferinde tüm kayıtları çekmeye çalışırsa:
- Sunucu tarafında: Sorgu süresi uzar, bellek tüketimi artar, diğer isteklere servis verme kapasitesi düşer.
- Ağ tarafında: Büyük response boyutları bant genişliğini tüketir, özellikle mobil istemciler için sorun yaratır.
- İstemci tarafında: Uygulama belleğe sığmayan veriyi işlemeye çalışırken çöker ya da donup kalır.
Pagination bu sorunu şu şekilde çözer: veriyi sayfalara böler, her sayfa için ayrı bir istek yapılmasına olanak tanır. Kullanıcı veya script hangi sayfada olduğunu takip eder ve veri bitene kadar döngüyle ilerler.
Gerçek bir senaryo düşünelim: Bir e-ticaret platformunun sipariş geçmişini her gece bir veri ambarına aktaran bir ETL pipeline’ı yazıyorsunuz. Günde 50.000 yeni sipariş geldiğini varsayalım. Bunu tek request’te çekmeye çalışmak yerine, 100’erli sayfalarla 500 istek atarak aynı veriyi çok daha güvenli ve kontrollü bir şekilde toplayabilirsiniz.
Pagination Türleri
Offset-Based Pagination (Sayfa Numarası ile)
En yaygın ve anlaşılması en kolay yöntemdir. page ve per_page (veya limit/offset) parametreleriyle çalışır.
- page: Hangi sayfada olduğunuzu belirtir
- per_page: Her sayfada kaç kayıt istediğinizi belirtir
- offset: Kaç kaydı atlayacağınızı belirtir (offset = page * per_page)
Örnek istek yapısı şöyle görünür: GET /api/users?page=3&per_page=100
Bu yöntemin en büyük avantajı kullanım kolaylığıdır. Dezavantajı ise büyük offset değerlerinde veritabanı performansının dramatik biçimde düşmesidir. PostgreSQL’de OFFSET 100000 yazmak, veritabanının o 100.000 kaydı atlayabilmek için önce okuması gerektiği anlamına gelir.
Cursor-Based Pagination (İmleç ile)
Modern API’lerin tercih ettiği yöntemdir. Offset yerine, son görülen kaydın bir “imleçini” (cursor) kullanır. Bu imleç genellikle base64 encode edilmiş bir ID veya timestamp’tir.
- Büyük veri setlerinde tutarlı performans sağlar
- Veri ekleme/silinmesinde sayfa kayması olmaz
- Geriye gitmek zordur, tek yön akış için idealdir
GitHub, Twitter ve Stripe gibi büyük platformlar cursor pagination kullanır.
Keyset Pagination
Cursor pagination’ın bir türevi olan keyset pagination, sıralama alanını (genellikle id veya created_at) doğrudan filtre olarak kullanır: GET /api/orders?after_id=12345&limit=100
Link Header Pagination
RFC 5988 standardına dayanan bu yaklaşımda, sonraki/önceki sayfa URL’leri response’un Link header’ında döner. GitHub API’si bu yöntemi kullanır.
Python ile Offset Pagination
Gelin gerçek bir senaryo üzerinden ilerleyelim. Bir HR sisteminden tüm çalışan kayıtlarını çeken ve CSV’ye yazan bir script yazalım.
pip install requests pandas#!/usr/bin/env python3
"""
Offset-based pagination ile API verisi çekme
Senaryo: HR sisteminden tüm çalışan verilerini toplu çekme
"""
import requests
import pandas as pd
import time
import logging
from typing import Optional
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
API_BASE_URL = "https://api.example-hr.com/v1"
API_TOKEN = "your-api-token-here"
PAGE_SIZE = 100
MAX_RETRIES = 3
RETRY_DELAY = 2 # saniye
def fetch_page(page: int, per_page: int = PAGE_SIZE) -> Optional[dict]:
"""Belirtilen sayfayı API'den çeker, hata durumunda retry uygular."""
headers = {
"Authorization": f"Bearer {API_TOKEN}",
"Content-Type": "application/json"
}
params = {
"page": page,
"per_page": per_page,
"sort": "id",
"order": "asc"
}
for attempt in range(MAX_RETRIES):
try:
response = requests.get(
f"{API_BASE_URL}/employees",
headers=headers,
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
logger.warning(f"Sayfa {page}, deneme {attempt + 1}: Timeout, bekleniyor...")
time.sleep(RETRY_DELAY * (attempt + 1))
except requests.exceptions.HTTPError as e:
if response.status_code == 429:
# Rate limit aşıldı, Retry-After header'ına bak
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning(f"Rate limit! {retry_after} saniye bekleniyor...")
time.sleep(retry_after)
elif response.status_code >= 500:
logger.error(f"Sunucu hatası: {e}, yeniden deneniyor...")
time.sleep(RETRY_DELAY * (attempt + 1))
else:
logger.error(f"İstemci hatası: {e}")
raise
logger.error(f"Sayfa {page} için maksimum deneme sayısı aşıldı!")
return None
def fetch_all_employees() -> list:
"""Tüm çalışan kayıtlarını sayfalı olarak çeker."""
all_employees = []
page = 1
total_pages = None
logger.info("Veri çekme başlıyor...")
while True:
logger.info(f"Sayfa çekiliyor: {page}" +
(f"/{total_pages}" if total_pages else ""))
data = fetch_page(page)
if not data:
logger.error(f"Sayfa {page} alınamadı, durduruluyor.")
break
# Toplam sayfa sayısını ilk response'dan al
if total_pages is None:
total_records = data.get("meta", {}).get("total", 0)
total_pages = (total_records + PAGE_SIZE - 1) // PAGE_SIZE
logger.info(f"Toplam kayıt: {total_records}, Toplam sayfa: {total_pages}")
employees = data.get("data", [])
if not employees:
logger.info("Boş sayfa alındı, tüm veri çekildi.")
break
all_employees.extend(employees)
logger.info(f"Sayfa {page} çekildi, {len(employees)} kayıt eklendi. "
f"Toplam: {len(all_employees)}")
# Son sayfaya ulaşıldı mı?
if total_pages and page >= total_pages:
break
page += 1
# Sunucuya nazik olmak için kısa bir bekleme
time.sleep(0.1)
return all_employees
if __name__ == "__main__":
employees = fetch_all_employees()
if employees:
df = pd.DataFrame(employees)
output_file = "employees_export.csv"
df.to_csv(output_file, index=False, encoding="utf-8-sig")
logger.info(f"Toplam {len(employees)} kayıt '{output_file}' dosyasına yazıldı.")
else:
logger.error("Hiçbir veri çekilemedi!")Bu scriptte dikkat etmemiz gereken birkaç kritik nokta var: exponential backoff ile retry mekanizması, rate limit tespiti ve toplam sayfa hesaplaması.
Cursor-Based Pagination: GitHub API Örneği
GitHub’ın API’si her iki yöntemi de destekler ama cursor (Link header) pagination’ı daha verimlidir. Bir organizasyonun tüm repository’lerini çeken gerçek bir örnek:
#!/usr/bin/env python3
"""
Link Header (Cursor) pagination ile GitHub API kullanımı
Senaryo: Bir organizasyonun tüm repo istatistiklerini çekme
"""
import requests
import json
import re
from typing import Optional
GITHUB_TOKEN = "ghp_your_token_here"
ORG_NAME = "your-organization"
def parse_link_header(link_header: str) -> dict:
"""GitHub Link header'ını parse eder ve next/prev URL'lerini döner."""
links = {}
if not link_header:
return links
for part in link_header.split(","):
section = part.strip().split(";")
if len(section) == 2:
url = section[0].strip()[1:-1] # < ve > karakterlerini kaldır
rel = re.findall(r'rel="([^"]+)"', section[1])[0]
links[rel] = url
return links
def fetch_org_repos(org_name: str) -> list:
"""Organizasyonun tüm repository'lerini cursor pagination ile çeker."""
headers = {
"Authorization": f"token {GITHUB_TOKEN}",
"Accept": "application/vnd.github.v3+json"
}
url = f"https://api.github.com/orgs/{org_name}/repos"
params = {
"per_page": 100,
"type": "all",
"sort": "updated"
}
all_repos = []
page_num = 0
while url:
page_num += 1
print(f"Sayfa {page_num} çekiliyor: {url}")
response = requests.get(url, headers=headers, params=params)
response.raise_for_status()
# Kalan rate limit bilgisini logla
remaining = response.headers.get("X-RateLimit-Remaining", "?")
reset_time = response.headers.get("X-RateLimit-Reset", "?")
print(f"Rate Limit Kalan: {remaining}")
repos = response.json()
all_repos.extend(repos)
print(f"Bu sayfada {len(repos)} repo, toplam: {len(all_repos)}")
# Link header'ından sonraki sayfayı al
link_header = response.headers.get("Link", "")
links = parse_link_header(link_header)
# Bir sonraki sayfanın URL'si (cursor)
url = links.get("next")
# İlk sayfadan sonra params'ı temizle (URL zaten parametreleri içeriyor)
params = {}
return all_repos
repos = fetch_org_repos(ORG_NAME)
print(f"nToplam {len(repos)} repository bulundu.")
# En yüksek star'lı 10 repo'yu göster
sorted_repos = sorted(repos, key=lambda x: x.get("stargazers_count", 0), reverse=True)
for repo in sorted_repos[:10]:
print(f" {repo['name']}: {repo['stargazers_count']} star")Bash ile Pagination: Hızlı ve Sade
Her zaman Python’a ihtiyaç yoktur. Basit bir veri dökümü için Bash yeterlidir:
#!/bin/bash
# Offset pagination ile API'den veri çeken Bash scripti
# Senaryo: Monitoring sisteminden alert geçmişini çekme
API_URL="https://monitoring.example.com/api/v2"
API_KEY="your-api-key"
OUTPUT_FILE="alerts_export.jsonl"
PAGE_SIZE=200
CURRENT_PAGE=1
TOTAL_FETCHED=0
# Çıktı dosyasını temizle
> "$OUTPUT_FILE"
echo "Alert geçmişi çekme başlıyor..."
while true; do
echo "Sayfa $CURRENT_PAGE çekiliyor..."
# API isteği at
RESPONSE=$(curl -s -w "n%{http_code}"
-H "Authorization: Bearer $API_KEY"
-H "Content-Type: application/json"
"${API_URL}/alerts?page=${CURRENT_PAGE}&per_page=${PAGE_SIZE}&status=resolved")
# HTTP status code'u ayır
HTTP_CODE=$(echo "$RESPONSE" | tail -n1)
BODY=$(echo "$RESPONSE" | head -n -1)
# Hata kontrolü
if [ "$HTTP_CODE" != "200" ]; then
echo "HATA: HTTP $HTTP_CODE alındı, durduruluyor."
echo "Response: $BODY"
exit 1
fi
# Kayıt sayısını al (jq gerekli)
RECORD_COUNT=$(echo "$BODY" | jq -r '.data | length')
TOTAL_PAGES=$(echo "$BODY" | jq -r '.meta.total_pages // "unknown"')
echo "Sayfa $CURRENT_PAGE/$TOTAL_PAGES: $RECORD_COUNT kayıt"
# Boş sayfa kontrolü
if [ "$RECORD_COUNT" -eq 0 ] 2>/dev/null; then
echo "Boş sayfa alındı. Tüm veri çekildi."
break
fi
# Her kaydı ayrı JSON satırı olarak yaz (JSONL format)
echo "$BODY" | jq -c '.data[]' >> "$OUTPUT_FILE"
TOTAL_FETCHED=$((TOTAL_FETCHED + RECORD_COUNT))
echo "Toplam çekilen: $TOTAL_FETCHED kayıt"
# Son sayfaya ulaşıldı mı?
HAS_NEXT=$(echo "$BODY" | jq -r '.meta.has_next_page // false')
if [ "$HAS_NEXT" = "false" ]; then
echo "Son sayfaya ulaşıldı."
break
fi
CURRENT_PAGE=$((CURRENT_PAGE + 1))
# Sunucuya biraz nefes ver
sleep 0.2
done
echo "Tamamlandı! Toplam $TOTAL_FETCHED kayıt '$OUTPUT_FILE' dosyasına yazıldı."
echo "Dosya boyutu: $(wc -l < "$OUTPUT_FILE") satır"Rate Limiting ile Profesyonel Yaklaşım
Üretim ortamında en sık karşılaşılan sorun rate limit’tir. Özellikle büyük veri setlerini toplu çekerken API sizi kısıtlamaya başlar. Bunu akıllıca yönetmek gerekir:
#!/usr/bin/env python3
"""
Akıllı rate limit yönetimi ile pagination
Senaryo: CRM'den müşteri verilerini çekme
"""
import requests
import time
import logging
from dataclasses import dataclass
from typing import Generator, Optional
logger = logging.getLogger(__name__)
@dataclass
class RateLimitConfig:
"""Rate limit konfigürasyon sınıfı."""
requests_per_minute: int = 60
burst_size: int = 10
backoff_factor: float = 1.5
max_backoff: int = 300 # maksimum bekleme süresi (saniye)
class PaginatedAPIClient:
"""Rate limit farkındalıklı sayfalama istemcisi."""
def __init__(self, base_url: str, api_key: str,
rate_config: Optional[RateLimitConfig] = None):
self.base_url = base_url
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"User-Agent": "DataExporter/1.0"
})
self.rate_config = rate_config or RateLimitConfig()
self.request_count = 0
self.start_time = time.time()
def _throttle(self):
"""İstek hızını kontrol eder, gerekirse bekler."""
self.request_count += 1
elapsed = time.time() - self.start_time
# Dakika başına istek oranını hesapla
if elapsed > 0:
current_rate = (self.request_count / elapsed) * 60
max_rate = self.rate_config.requests_per_minute
if current_rate > max_rate:
wait_time = (self.request_count / (max_rate / 60)) - elapsed
if wait_time > 0:
logger.debug(f"Throttle: {wait_time:.2f}s bekleniyor "
f"(mevcut oran: {current_rate:.1f}/dk)")
time.sleep(wait_time)
def _handle_rate_limit_response(self, response: requests.Response) -> float:
"""429 response'undan bekleme süresini hesaplar."""
# Retry-After header'ı varsa onu kullan
retry_after = response.headers.get("Retry-After")
if retry_after:
return float(retry_after)
# X-RateLimit-Reset header'ı varsa onu kullan
reset_time = response.headers.get("X-RateLimit-Reset")
if reset_time:
wait = float(reset_time) - time.time()
return max(wait, 1)
return 60 # Varsayılan: 1 dakika bekle
def paginate(self, endpoint: str, params: dict = None) -> Generator:
"""Endpoint'i sayfalı olarak gezer, kayıtları generator ile döner."""
params = params or {}
params.setdefault("per_page", 100)
page = 1
backoff = 1
while True:
params["page"] = page
self._throttle()
try:
response = self.session.get(
f"{self.base_url}{endpoint}",
params=params,
timeout=30
)
if response.status_code == 429:
wait_time = self._handle_rate_limit_response(response)
logger.warning(f"Rate limit! {wait_time:.0f}s bekleniyor...")
time.sleep(wait_time)
continue # Aynı sayfayı tekrar dene
response.raise_for_status()
backoff = 1 # Başarılı istek, backoff'u sıfırla
data = response.json()
records = data.get("data", data if isinstance(data, list) else [])
if not records:
logger.info(f"Sayfalama tamamlandı. Toplam {page-1} sayfa işlendi.")
break
logger.info(f"Sayfa {page}: {len(records)} kayıt döndü")
yield from records
# Sonraki sayfa var mı?
meta = data.get("meta", {})
if not meta.get("has_next_page", len(records) == params["per_page"]):
break
page += 1
except requests.exceptions.RequestException as e:
logger.error(f"İstek hatası (sayfa {page}): {e}")
wait_time = min(backoff * self.rate_config.backoff_factor,
self.rate_config.max_backoff)
logger.info(f"{wait_time:.0f}s sonra yeniden deneniyor...")
time.sleep(wait_time)
backoff = wait_time
# Kullanım örneği
client = PaginatedAPIClient(
base_url="https://api.crm.example.com/v1",
api_key="your-crm-api-key",
rate_config=RateLimitConfig(requests_per_minute=30)
)
all_customers = []
for customer in client.paginate("/customers", params={"country": "TR", "active": True}):
all_customers.append(customer)
if len(all_customers) % 1000 == 0:
print(f"{len(all_customers)} müşteri işlendi...")
print(f"Toplam {len(all_customers)} müşteri çekildi.")Paralel Pagination: İleri Seviye Hız Optimizasyonu
Sayfa sayısını önceden bildiğiniz durumlarda, sayfaları paralel olarak çekebilirsiniz. Bu, toplam çekme süresini önemli ölçüde kısaltır. Ancak rate limit’e dikkat etmek gerekir:
#!/usr/bin/env python3
"""
ThreadPoolExecutor ile paralel pagination
DİKKAT: Sadece API rate limit'i uygunsa kullanın!
Senaryo: Büyük ürün kataloğunu hızlıca indirme
"""
import requests
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from threading import Lock
API_URL = "https://api.catalog.example.com/v1"
API_TOKEN = "your-token"
MAX_WORKERS = 5 # Aynı anda maksimum 5 paralel istek
PAGE_SIZE = 200
results_lock = Lock()
all_products = []
def get_total_pages() -> int:
"""İlk isteği atarak toplam sayfa sayısını öğren."""
response = requests.get(
f"{API_URL}/products",
headers={"Authorization": f"Bearer {API_TOKEN}"},
params={"page": 1, "per_page": PAGE_SIZE}
)
response.raise_for_status()
data = response.json()
total = data["meta"]["total_records"]
return (total + PAGE_SIZE - 1) // PAGE_SIZE
def fetch_page_parallel(page_num: int) -> tuple:
"""Tek bir sayfayı çeker, (page_num, records) tuple'ı döner."""
try:
response = requests.get(
f"{API_URL}/products",
headers={"Authorization": f"Bearer {API_TOKEN}"},
params={"page": page_num, "per_page": PAGE_SIZE},
timeout=30
)
response.raise_for_status()
data = response.json()
return (page_num, data.get("data", []))
except Exception as e:
print(f"Sayfa {page_num} hatası: {e}")
return (page_num, [])
def fetch_all_parallel():
"""Tüm sayfaları paralel olarak çeker."""
print("Toplam sayfa sayısı alınıyor...")
total_pages = get_total_pages()
print(f"Toplam {total_pages} sayfa paralel çekilecek ({MAX_WORKERS} worker ile)")
ordered_results = {}
completed = 0
with ThreadPoolExecutor(max_workers=MAX_WORKERS) as executor:
future_to_page = {
executor.submit(fetch_page_parallel, page): page
for page in range(1, total_pages + 1)
}
for future in as_completed(future_to_page):
page_num, records = future.result()
ordered_results[page_num] = records
completed += 1
if completed % 10 == 0:
print(f"İlerleme: {completed}/{total_pages} sayfa tamamlandı")
# Worker'lar arası küçük bir bekleme (rate limit koruması)
time.sleep(0.05)
# Sayfaları sıralı birleştir
final_results = []
for page_num in sorted(ordered_results.keys()):
final_results.extend(ordered_results[page_num])
return final_results
products = fetch_all_parallel()
print(f"Toplam {len(products)} ürün çekildi.")Gerçek Dünya Sorunları ve Çözümleri
Sayfalama ile çalışırken production’da sık karşılaşılan birkaç tuzaktan bahsedelim.
Veri kayması problemi: Offset-based pagination kullanırken, siz sayfalarda gezinirken yeni kayıtlar eklenirse veya silinirse, bazı kayıtları atlayabilir ya da iki kez görebilirsiniz. Çözüm, çekme başlamadan önce bir snapshot_time belirlemek ve tüm isteklerde bu filtreyi kullanmaktır: ?created_before=2024-01-15T10:00:00Z&page=X
Büyük offset sorunları: Bir API’nin 500. sayfasını çekmeye çalıştığınızda yanıt süresinin aniden uzadığını fark edebilirsiniz. Bu, veritabanının OFFSET 50000 yapması gerektiği anlamına gelir. Eğer API cursor pagination desteklemiyorsa ve bunu kontrol edemiyorsanız, veriyi daha küçük zaman dilimlerine bölerek çekin.
Tutarsız total_count: Bazı API’ler her sayfada farklı total değeri döndürebilir (çok nadir ama yaşanıyor). Scripti her zaman hem sayfa numarasına hem de “boş sayfa” kontrolüne güvenecek şekilde yazın, sadece total_pages‘e güvenmeyin.
Yarım kalan işlemler: Büyük bir çekim işlemi yarıda kesilirse baştan başlamak istemezsiniz. Checkpoint mekanizması ekleyin: her 1000 kayıtta bir ilerlemeyi bir dosyaya kaydedin ve script yeniden başladığında kaldığı yerden devam etsin.
Sonuç
Pagination, API entegrasyonlarının bel kemiğidir ve doğru uygulanmadığında hem production sistemlerinizi hem de entegre olduğunuz API’yi gereksiz yere yorar. Temel prensipleri bir daha sıralayalım:
- Offset pagination ile basit ve küçük veri setlerinde kullanımı kolaydır, büyük veri setlerinde performans sorunu yaratır.
- Cursor pagination büyük ve sürekli büyüyen veri setleri için doğru tercihtir.
- Her zaman retry mekanizması ve rate limit yönetimi ekleyin, bunlar isteğe bağlı değil zorunludur.
- Paralel pagination hız kazandırır ama rate limit’e saygı göstererek yapılmalıdır.
- Uzun çekimler için checkpoint mekanizması, yarıda kalan işlerin baştan başlamasını önler.
Production’da bu teknikleri uygularken her zaman kendinize şu soruyu sorun: “Bu script gece 3’te başarısız olursa ve ben uyuyorsam, sabah uyandığımda ne kadar veri kaybedilmiş olur?” Cevap “hiçbiri” ise doğru yazmışsınız demektir.