Otomasyonun Faturası: İş Başına Maliyeti Ölçmek ve Bütçe Alarmı Kurmak

Otomasyon maliyeti çoğu küçük ekipte tek bir satırda görünür: ay sonunda gelen bulut faturası. Sunucu kirası, API çağrıları, depolama ve dil modeli kullanımı aynı toplamın içinde erir; hangi işin ne kadar harcadığı bilinmez. Bilinmeyince de karar verilemez: fiyat güncelleme betiğini saatte bir yerine günde bir çalıştırmak ne kazandırır, modelli özetleme adımı gerçekten değer mi, bunların cevabı yoktur.

Çözüm büyük bir maliyet yönetimi platformu değil, küçük bir defterdir. Her koşum bittiğinde kendi giderini yazar: hangi iş, ne kadar sürdü, kaç API çağrısı yaptı, kaç token tüketti. Defterin üstüne bir aylık toplam sorgusu ve bir eşik kontrolü eklenince, bütçe aşımı ay sonunda değil aşıldığı gün görünür.

Bu yazıda defteri SQLite ile kuruyor, koşumu saran bir kayıt bloğu yazıyor ve bütçeyi geçen işi zamanlanmış zincirden düşüren bir alarm betiği ekliyoruz. Örnekler Python ve standart kütüphaneyle çalışır; dış bağımlılık yoktur.

Maliyet defteri nedir ve neyi kaydeder

Maliyet defteri, her otomasyon koşumu için bir satır ve o koşuma ait gider kalemleri için birer alt satır tutan küçük bir veri tabanıdır. Koşum satırı işin adını, başlangıç zamanını, süresini ve sonucunu taşır. Gider satırları ise kalem adı, miktar, birim fiyat ve tutarı içerir; böylece fiyat değişse bile eski koşumların maliyeti olduğu gibi kalır.

Defterin iki tablodan oluşmasının nedeni, giderlerin türden türe farklı ölçülmesidir. Sunucu maliyeti saniyeyle, API maliyeti çağrı sayısıyla, model maliyeti token sayısıyla, depolama ise yazılan megabaytla ölçülür. Hepsini tek tabloya sıkıştırmak yerine kalemleri satır olarak tutmak, yeni bir gider türü eklendiğinde şema değiştirmeyi gerektirmez.

Tutarın koşum anında hesaplanıp yazılması bilinçli bir tercihtir. Birim fiyatlar zamanla değişir; sonradan hesaplanan bir rapor, geçmiş ayı bugünün fiyatıyla yeniden fiyatlandırır ve yanıltır. Deftere yazılan tutar, o günkü fiyatla o günkü tüketimi kaydeder ve muhasebe kaydı gibi değişmez.

Birim fiyat tablosunu doldurmak

Birim fiyat tablosu, her gider kaleminin tek biriminin kaça mal olduğunu tutan bir sözlüktür ve kaynağı sağlayıcının kendi fiyat listesidir. Sunucu için aylık ücret saniyeye bölünür, API için sağlayıcının çağrı ya da istek başı fiyatı alınır, model için token başı fiyat yazılır. Tahmin değil, fiyat listesinden okunan sayı girilir.

Sunucu payını dağıtırken basit bir yöntem yeterlidir: aylık sunucu ücreti ayın saniye sayısına bölünür ve koşum süresiyle çarpılır. Sunucuda tek iş çalışmıyorsa bu yöntem yalnız süreyi paylaştırır, işlemci yükünü değil. Küçük ekipler için bu yaklaşım kabul edilebilir bir başlangıçtır; daha hassas dağıtım gerekince süre yerine işlemci saniyesi ölçülür.

Tabloyu doldururken şu kalemler çoğu otomasyonu kapsar:

• sunucu saniyesi: aylık kira bölü ayın saniye sayısı
• API çağrısı: sağlayıcının istek başı fiyatı; kotalı planlarda plan ücreti bölü kota
• model tokeni: giriş ve çıkış için ayrı fiyat varsa iki ayrı kalem
• depolama: yazılan megabayt başına aylık ücret
• e-posta ya da SMS: gönderim başı fiyat

SQLite defteri ve koşumu saran kayıt bloğu

SQLite defteri, işin çalıştığı makinede tek bir dosya olarak durur ve ayrı bir servis gerektirmez. Kayıt bloğu bir bağlam yöneticisidir: işin gövdesi bloğun içinde çalışır, blok süreyi ölçer, sayaçları toplar ve çıkışta hem koşum satırını hem gider satırlarını yazar. İş hata verse bile kayıt düşer, çünkü başarısız koşum da para harcamıştır.

SQLite'ın tek dosyada kilit ve işlem güvencesi sunması, bu tür küçük defterler için yeterlidir; ayrıntısı SQLite ile küçük ama sağlam veri boru hatları yazısında anlatılmıştı. Aşağıdaki kod tabloları oluşturur, kayıt bloğunu tanımlar ve örnek bir koşumu deftere işler.

import sqlite3
import time
from contextlib import contextmanager
from datetime import datetime, timezone

DB = "maliyet.db"

# Sağlayıcının fiyat listesinden alınır; buradakiler yer tutucu.
BIRIM_FIYAT = {
    "sunucu_saniye": 0.000004,   # aylık sunucu ücreti / (30 * 86400)
    "api_cagri": 0.0005,
    "model_token": 0.000003,
    "depolama_mb": 0.00002,
}


def hazirla(bag):
    bag.execute("""
        CREATE TABLE IF NOT EXISTS kosum (
            id INTEGER PRIMARY KEY,
            is_adi TEXT NOT NULL,
            baslangic TEXT NOT NULL,
            sure_sn REAL NOT NULL,
            durum TEXT NOT NULL
        )""")
    bag.execute("""
        CREATE TABLE IF NOT EXISTS gider (
            kosum_id INTEGER NOT NULL REFERENCES kosum(id),
            kalem TEXT NOT NULL,
            miktar REAL NOT NULL,
            birim_fiyat REAL NOT NULL,
            tutar REAL NOT NULL
        )""")


class Sayac:
    def __init__(self):
        self.miktar = {}

    def ekle(self, kalem, adet=1):
        self.miktar[kalem] = self.miktar.get(kalem, 0) + adet


@contextmanager
def maliyet_kaydi(is_adi):
    sayac = Sayac()
    baslangic = datetime.now(timezone.utc).isoformat(timespec="seconds")
    t0 = time.perf_counter()
    durum = "basarili"
    try:
        yield sayac
    except Exception:
        durum = "hatali"
        raise
    finally:
        sure = time.perf_counter() - t0
        sayac.ekle("sunucu_saniye", sure)
        with sqlite3.connect(DB) as bag:
            hazirla(bag)
            kosum_id = bag.execute(
                "INSERT INTO kosum (is_adi, baslangic, sure_sn, durum) "
                "VALUES (?, ?, ?, ?)",
                (is_adi, baslangic, sure, durum),
            ).lastrowid
            for kalem, miktar in sayac.miktar.items():
                fiyat = BIRIM_FIYAT[kalem]
                bag.execute(
                    "INSERT INTO gider VALUES (?, ?, ?, ?, ?)",
                    (kosum_id, kalem, miktar, fiyat, miktar * fiyat),
                )


if __name__ == "__main__":
    with maliyet_kaydi("fiyat-guncelleme") as s:
        for _ in range(120):
            s.ekle("api_cagri")            # her gerçek çağrıdan sonra
        s.ekle("model_token", 18500)       # yanıttaki usage alanından
        s.ekle("depolama_mb", 3.2)

Sayaç değerleri tahmin edilmez, ölçülür: API istemcisi her isteğin ardından bir çağrı ekler, model yanıtındaki kullanım alanından token sayısı okunur, dosya yazıldıktan sonra boyutu alınır. Kayıt bloğu bu sayaçları fiyatla çarpıp yazar; işin kendisi maliyetle ilgili hiçbir hesap yapmaz.

Aylık bütçe alarmı kurmak

Bütçe alarmı, defterdeki ay içi toplamı her iş için tanımlı üst sınırla karşılaştıran ve sınır aşıldığında sıfırdan farklı bir çıkış koduyla biten küçük bir betiktir. Zamanlanmış zincirin başına konur: bütçe betiği geçerse asıl iş çalışır, geçmezse zincir orada durur. Böylece aşım, faturada değil bir sonraki koşumdan önce yakalanır.

Betik iki şey üretir: iş başına aylık özet tablosu ve aşım listesi. Özet tablo koşum sayısı, toplam tutar ve koşum başı ortalamayı gösterir; koşum başı maliyetteki ani artış, bütçe dolmadan önce bir sorunun işaretidir. Örneğin sayfalama döngüsü bozulup aynı API'yi binlerce kez çağıran bir iş, toplam bütçeyi aşmadan önce koşum başı sütununda görünür.

import sqlite3
import sys
from datetime import datetime, timezone

DB = "maliyet.db"
AYLIK_BUTCE = {"fiyat-guncelleme": 25.0, "gunluk-rapor": 10.0}   # USD
UYARI_ORANI = 0.8   # bütçenin bu kadarı dolunca uyarı yaz

SORGU = """
SELECT k.is_adi,
       COUNT(DISTINCT k.id)                              AS kosum,
       ROUND(SUM(g.tutar), 4)                            AS toplam,
       ROUND(SUM(g.tutar) / COUNT(DISTINCT k.id), 5)     AS kosum_basi
FROM kosum k JOIN gider g ON g.kosum_id = k.id
WHERE substr(k.baslangic, 1, 7) = ?
GROUP BY k.is_adi
ORDER BY toplam DESC
"""


def rapor(ay):
    with sqlite3.connect(DB) as bag:
        return bag.execute(SORGU, (ay,)).fetchall()


def main():
    ay = datetime.now(timezone.utc).strftime("%Y-%m")
    asim, uyari = [], []
    for is_adi, kosum, toplam, kosum_basi in rapor(ay):
        butce = AYLIK_BUTCE.get(is_adi)
        durum = "bütçe tanımsız"
        if butce is not None:
            oran = toplam / butce
            durum = "bütçenin %.0f%%'i doldu" % (oran * 100)
            if oran >= 1:
                asim.append(is_adi)
            elif oran >= UYARI_ORANI:
                uyari.append(is_adi)
        print("%-20s %5d koşum  %9.4f USD  koşum başı %.5f  %s"
              % (is_adi, kosum, toplam, kosum_basi, durum))
    if uyari:
        print("UYARI:", ", ".join(uyari), file=sys.stderr)
    if asim:
        print("BÜTÇE AŞILDI:", ", ".join(asim), file=sys.stderr)
        sys.exit(2)


if __name__ == "__main__":
    main()

Zamanlayıcıda kullanım tek satırdır: python3 butce.py && python3 fiyat_guncelleme.py. Bütçe betiği çıkış kodu 2 ile bitince kabuk ikinci komutu çalıştırmaz. Aynı zincire kilit ve yeniden çalıştırma mantığı da ekleniyorsa, sıralama için zamanlanmış iş tasarımı yazısındaki düzen uygundur: önce bütçe, sonra kilit, sonra iş.

Alarm eşiği bütçenin tamamı olmak zorunda değildir. Uyarı oranı, aşım gerçekleşmeden önce e-posta ya da mesaj gönderilecek eşiği belirler; asıl iş bu aşamada durmaz. Aşım ise işi durdurur ve durdurma kararı bilinçli olarak sert tutulur: bütçesiz devam eden otomasyon, kimsenin bakmadığı bir saatte en pahalı hatayı yapar.

Sonuç: ilk adım tek bir işi deftere bağlamak

Otomasyon maliyeti, her koşumun kendi giderini yazdığı bir defter olmadan yönetilemez; fatura toplamı hangi işin ne harcadığını söylemez. Defter iki SQLite tablosu, bir bağlam yöneticisi ve bir bütçe betiğinden ibarettir. Fiyatlar sağlayıcının listesinden alınır, sayaçlar ölçülür, tutar koşum anında hesaplanıp değişmez şekilde kaydedilir.

İlk adım olarak en sık çalışan ya da en çok dış çağrı yapan tek bir işi seçip kayıt bloğunun içine alın. Bir hafta sonra aylık özet sorgusu o iş için koşum başı maliyeti gösterir; bütçe sınırını bu sayıya bakarak koyun, tahminle değil. Sonra diğer işleri sırayla ekleyin.

Elinizdeki otomasyonların hangi kalemlerden para harcadığını çıkarmak ve bütçe alarmını kendi zincirinize yerleştirmek için iletişim bölümünden yazabilirsiniz.