Climatiq API: Integrujte výpočty emisí GHG do backendu
V roce 2025 se trh s řešeními pro monitorování emisí uhlíku rozrostl dokud 17,3 miliardy dolarůřízená stále přísnějšími předpisy jako je evropská CSRD, klimatické zveřejnění SEC v USA a ISO 14064. Společnosti nemohou už se nemusíte spoléhat na excelové listy a hrubé odhady: jsou užitečné automatizované výpočty, ověřitelné a integrované do operačních systémů.
Technická výzva je skutečná: výpočet emisí skleníkových plynů vyžaduje přístup k databázím emisní faktory aktualizované, metodiky ověřené protokolem GHG a konverze mezi stovkami různých měrných jednotek. Budování a údržba toho všeho v domě vyžaduje měsíce specializované práce. Tady to přichází na řadu Climatiq.
Climatiq je REST API, které poskytuje přístup k over 190 000 emisních faktorů z více než 40 ověřených zdrojů (EPA, DEFRA/BEIS, IEA, ecoinvent), pokrytí 300+ geografických regionůa výpočty v souladu s protokolem GHG pro rozsah 1, 2 a 3. V tomto článku postavíme a Kompletní backend FastAPI která integruje Climatiq pro výpočty emisí z výroby spolu s klientem TypeScript a kalkulačkou v reálném čase pro aplikace SaaS.
Co se naučíte
- Architektura API Climatiq: koncové body, autentizace, limity rychlosti a datové modely
- Databáze emisních faktorů: jak vyhledávat, filtrovat a vybírat správné faktory
- Odhad na základě činností: výpočty založené na konkrétních činnostech (kWh, km, kg)
- Odhad na základě útraty: výpočty měnových výdajů, když chybí primární data
- Rozsah protokolu GHG 1, 2, 3: mapování kategorií a vyhovující výpočty
- Robustní klient Pythonu s opakováním, ukládáním do mezipaměti Redis a zpracováním chyb
- Klient TypeScript/Node.js s Axios a typy pro integraci frontendu
- Kalkulačka uhlíku v reálném čase pro SaaS s uhlíkovými štítky na produktech
- Testování pomocí simulovaného API pro prostředí CI/CD
- Alternativy ke Climatiq a porovnání funkčnosti
Zelená řada softwarového inženýrství
Tento článek je součástí kompletní série o Green Software Engineering. Každá položka řeší konkrétní aspekt digitální udržitelnosti:
| # | Položka | Hlavní téma |
|---|---|---|
| 1 | Principy zeleného softwarového inženýrství | GSF, SCI spec, 8 základních principů |
| 2 | CodeCarbon: Kód měření emisí | Knihovna Pythonu, dashboard, integrace CI/CD |
| 3 | Climatiq API: Výpočty skleníkových plynů v backendu | REST API, rozsah 1-3, integrace FastAPI + TypeScript |
| 4 | Carbon Aware SDK | Přesouvání pracovní zátěže, intenzita mřížky, časový posun |
| 5 | Rozsah 1, 2 a 3: Modelování dat pro ESG Reporting | Struktura dat, výpočty, agregace, reporting |
| 6 | GreenOps: Carbon-Aware Infrastructure | Plánování Kubernetes, škálování na základě problémů |
| 7 | Hodnotový řetězec emisního potrubí Rozsah 3 | Sběr dodavatelských dat, kalkulace, audit trail |
| 8 | ESG Reporting API: Integrace CSRD | CSRD workflow, automatizace reportů, compliance |
| 9 | Udržitelné architektonické vzory | Úložiště, inteligentní ukládání do mezipaměti, dávka s ohledem na uhlík |
| 10 | AI a Carbon: ML Training Footprint | LLM emise, optimalizace, zelená AI |
1. Protokol GHG a potřeba automatizovaných výpočtů
Il GHG Protocol Corporate Standard je to nejrozšířenější rámec na světě pro účtování firemních emisí. Klasifikuje emise do tří oblastí:
- Rozsah 1 (přímé emise): Spalování pohonných hmot ve služebních vozidlech, výrobní závody, vytápění. Jsou pod přímou kontrolou společnosti.
- Rozsah 2 (Nepřímá energie): Nakoupena elektřina, pára, teplo. Dělí se na na základě polohy (místní síťový mix) e tržní (energetické certifikáty, PPA).
- Rozsah 3 (hodnotový řetězec): 15 kategorií, které zahrnují nákup zboží a služby, doprava po směru/po proudu, použití produktu, konec životnosti, služební cesty, dojíždění zaměstnanců a mnoho dalšího. Obvykle představují 70–90 % celkových emisí společnosti.
Pro každý výpočet potřebujete a emisní faktor: koeficient, který převádí činnost (litry nafty, kWh elektřiny, nákup v eurech) v kg CO₂e. Tyto faktory se liší podle:
- Referenční rok: elektrické sítě se každý rok mění
- Zeměpisná oblast: Italská kWh se liší od německé kWh
- Zdroj dat: EPA pro USA, DEFRA pro Velkou Británii, ISPRA pro Itálii
- Obvod: proti proudu, po proudu, od kolébky po bránu, od kolébky po hrob
Udržování aktualizované databáze emisních faktorů vyžaduje specializovaný tým. Climatiq to dělá za nás a dále agreguje 40 ověřených zdrojů s průběžné aktualizace.
2. Přehled Climatiq API
Architektura a hlavní koncové body
Climatiq API je JSON REST API založené na URL https://beta3.api.climatiq.io.
Autentizace se provádí přes Token na doručitele v hlavičce HTTP.
Dostupné plány zahrnují:
| Patro | Volání/měsíc | Funkčnost | Typické použití |
|---|---|---|---|
| Společenství | 250 | Všechny koncové body | Prototypování, testování |
| Startér | 5 000 | + Soukromé faktory | MSP, MVP |
| Růst | 50 000 | + SLA, podpora | Rostoucí společnosti |
| Podnik | Zvyk | + Audit trail, SSO | Velké organizace |
Hlavní koncové body jsou:
| Koncové body | Metoda | Popis |
|---|---|---|
/search |
ZÍSKAT | Vyhledávání emisních faktorů v databázi |
/estimate |
ZVEŘEJNIT | Odhadovaná jednotlivá emise z činností |
/batch/estimate |
ZVEŘEJNIT | Více odhadů (až 100 na žádost) |
/travel/flights |
ZVEŘEJNIT | Emise z letecké společnosti (rozsah 3.6) |
/freight |
ZVEŘEJNIT | Emise z multimodální nákladní dopravy |
/procurement |
ZVEŘEJNIT | Problémy s nákupem (rozsah 3.1, podle útraty) |
/energy |
ZVEŘEJNIT | Emise spotřeby energie (rozsah 2) |
/compute |
ZVEŘEJNIT | Emise cloud computingu |
Struktura žádosti o odhad
# Esempio di richiesta POST /estimate
{
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"data_version": "^21",
"region": "IT"
},
"parameters": {
"energy": 1000,
"energy_unit": "kWh"
}
}
# Risposta
{
"co2e": 0.415,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"co2e_calculation_origin": "source",
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"source": "IEA",
"year": 2022,
"region": "IT",
"category": "Electricity",
"lca_activity": "electricity_generation",
"data_quality_flags": []
},
"constituent_gases": {
"co2e_total": 0.415,
"co2e_other": null,
"co2": 0.415,
"ch4": null,
"n2o": null
}
}Verze dat: Sémantika a doporučené postupy
Pole data_version řídí, jakou verzi databáze použít.
Stříška (^21) použijte kompatibilní verzi 21 nebo vyšší, zajistěte
automatické aktualizace emisních faktorů. ve výrobě, blokovat
přesnou verzi (např. "21") pro reprodukovatelnost
výpočty a auditní záznamy. Záměrně upgradujte na nové verze softwaru
s explicitním přepočtem historických hodnot.
3. Datový model: Databáze emisních faktorů
Struktura emisního faktoru
Emisní faktor v Climatiq je jednoznačně identifikován
activity_id, source, region, year
e lca_activity. Pochopení této struktury je zásadní
pro výběr správného faktoru.
# Struttura di un Emission Factor nel database Climatiq
{
"activity_id": "fuel_type-diesel",
"uuid": "94de5038-8b06-4e24-8e8c-1b87e1e0",
"name": "Diesel",
"category": "Fuel",
"sector": "Transport",
"source": "DEFRA",
"source_link": "https://www.gov.uk/guidance/ghg-conversion-factors-for-company-reporting",
"source_dataset": "DEFRA 2023",
"year": 2023,
"year_released": 2023,
"region": "GB",
"region_name": "United Kingdom",
"description": "Diesel combustion emission factor",
"unit_type": ["Volume", "Weight"],
"supported_calculation_methods": ["ar5"],
"factor": 2.5179,
"factor_calculation_method": "ar5",
"factor_calculation_origin": "source",
"constituent_gases": {
"co2e_total": 2.5179,
"co2": 2.5148,
"ch4": 0.0009,
"n2o": 0.0022
}
}Hlavní zdroje dat
| Zdroj | Země/oblast | Aktualizovat | Pokryté sektory |
|---|---|---|---|
| DEFRA/BEIS | UK | Výroční | Energie, doprava, nákupy, materiál |
| EPA | USA | Výroční | Energetika, průmyslové procesy, zemědělství |
| IEA | Globální | Výroční | Elektřina podle zemí, primární energie |
| ekoinvent | Globální | Pololetně | Kompletní LCA, dodavatelský řetězec, materiály |
| ADEME | Francie | Výroční | Uhlí, doprava, FR energetika |
| EHP | EU | Výroční | Síťová elektřina Evropa, sektorové emise |
| ISPRA | Itálie | Výroční | Národní inventura skleníkových plynů, mix elektřiny IT |
4. Odhad na základě činností: Výpočty pro konkrétní činnosti
Přístup na základě činnosti je nejpřesnější: je založen na údaje o primární činnosti, jako jsou litry spotřebovaného paliva, kWh elektřiny, ujeté kilometry nebo tuny nakoupeného materiálu. Vyžaduje sběr provozních dat podrobné, ale poskytuje vědecky spolehlivé výsledky.
Příklad: Výpočet emisí z vozového parku společnosti (rozsah 1)
# activity_based_estimation.py
import httpx
from dataclasses import dataclass
from enum import Enum
class FuelType(str, Enum):
DIESEL = "fuel_type-diesel"
PETROL = "fuel_type-petrol"
HVO = "fuel_type-hvo_biodiesel"
LPG = "fuel_type-lpg"
CNG = "fuel_type-cng"
@dataclass
class VehicleTrip:
vehicle_id: str
fuel_type: FuelType
litres_consumed: float
region: str = "IT"
async def estimate_fleet_scope1(
api_key: str,
trips: list[VehicleTrip],
data_version: str = "^21"
) -> dict:
"""
Calcola Scope 1 per flotta aziendale con activity-based estimation.
Returns aggregato e dettaglio per veicolo.
"""
BASE_URL = "https://beta3.api.climatiq.io"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Costruisce le richieste batch (max 100 per chiamata)
batch_requests = [
{
"emission_factor": {
"activity_id": trip.fuel_type.value,
"data_version": data_version,
"region": trip.region
},
"parameters": {
"volume": trip.litres_consumed,
"volume_unit": "l"
}
}
for trip in trips
]
async with httpx.AsyncClient(timeout=30.0) as client:
response = await client.post(
f"{BASE_URL}/batch/estimate",
headers=headers,
json={"requests": batch_requests}
)
response.raise_for_status()
results = response.json()["results"]
# Aggrega risultati per veicolo
vehicle_emissions = {}
total_co2e_kg = 0.0
for trip, result in zip(trips, results):
if "error" in result:
print(f"Errore per veicolo {trip.vehicle_id}: {result['error']}")
continue
co2e_kg = result["co2e"]
total_co2e_kg += co2e_kg
if trip.vehicle_id not in vehicle_emissions:
vehicle_emissions[trip.vehicle_id] = {
"total_co2e_kg": 0.0,
"fuel_type": trip.fuel_type.value,
"trips": 0
}
vehicle_emissions[trip.vehicle_id]["total_co2e_kg"] += co2e_kg
vehicle_emissions[trip.vehicle_id]["trips"] += 1
return {
"scope": "scope_1",
"total_co2e_kg": total_co2e_kg,
"total_co2e_tco2e": total_co2e_kg / 1000,
"vehicles": vehicle_emissions,
"vehicle_count": len(vehicle_emissions)
}
# Utilizzo
import asyncio
async def main():
trips = [
VehicleTrip("VAN-001", FuelType.DIESEL, 120.5),
VehicleTrip("VAN-002", FuelType.DIESEL, 98.3),
VehicleTrip("TRUCK-001", FuelType.HVO, 245.0),
VehicleTrip("CAR-001", FuelType.PETROL, 42.1),
]
result = await estimate_fleet_scope1(
api_key="clq_live_your_key_here",
trips=trips
)
print(f"Scope 1 totale: {result['total_co2e_tco2e']:.2f} tCO2e")
for vid, data in result["vehicles"].items():
print(f" {vid}: {data['total_co2e_kg']:.1f} kg CO2e")
asyncio.run(main())Výpočet emisí elektřiny (rozsah 2)
Rozsah 2 vyžaduje výpočet dvěma metodami: na základě umístění (síťový mix) a tržní (zbytkový mix faktor nebo PPA). Obě hodnoty musí být uvedeny v zveřejnění protokolu GHG.
# scope2_electricity.py
from enum import Enum
class Scope2Method(str, Enum):
LOCATION_BASED = "location_based"
MARKET_BASED = "market_based"
async def estimate_scope2_electricity(
client: httpx.AsyncClient,
api_key: str,
kwh_consumed: float,
region: str,
method: Scope2Method,
renewable_percentage: float = 0.0,
data_version: str = "^21"
) -> dict:
"""
Scope 2 con metodo location-based o market-based.
- location_based: usa il mix elettrico della rete locale
- market_based: usa residual mix o certificati (IRECs, GOs)
"""
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
# Market-based: se 100% rinnovabile certificato, le emissioni sono zero
if method == Scope2Method.MARKET_BASED and renewable_percentage >= 100.0:
return {
"co2e_kg": 0.0,
"co2e_tco2e": 0.0,
"method": method.value,
"note": "100% energie rinnovabili certificate - emissioni zero"
}
# Activity ID diverso per location-based vs market-based
if method == Scope2Method.LOCATION_BASED:
activity_id = "electricity-supply_grid-source_residual_mix"
else:
# Market-based usa residual mix (esclude RES certificate)
activity_id = "electricity-supply_grid-source_residual_mix"
# Riduzione proporzionale per rinnovabili parziali (market-based)
effective_kwh = kwh_consumed
if method == Scope2Method.MARKET_BASED and renewable_percentage > 0:
effective_kwh = kwh_consumed * (1 - renewable_percentage / 100.0)
payload = {
"emission_factor": {
"activity_id": activity_id,
"data_version": data_version,
"region": region
},
"parameters": {
"energy": effective_kwh,
"energy_unit": "kWh"
}
}
response = await client.post(
"https://beta3.api.climatiq.io/estimate",
headers=headers,
json=payload
)
response.raise_for_status()
data = response.json()
return {
"co2e_kg": data["co2e"],
"co2e_tco2e": data["co2e"] / 1000,
"co2e_unit": data["co2e_unit"],
"method": method.value,
"kwh_consumed": kwh_consumed,
"effective_kwh": effective_kwh,
"renewable_percentage": renewable_percentage,
"emission_factor": data["emission_factor"],
"region": region
}5. Odhad na základě útraty: Výpočty z údajů o útratě
Když nemáte přesné údaje o aktivitě, přístup založené na výdajích používá měnové výdaje jako proxy pro aktivitu. Climatiq uplatňuje emisní faktory ekonomické (kg CO₂e na euro/utracený dolar) podle kategorie nákupu na základě údajů vstup-výstup tabulek OECD nebo EXIOBASE. Je to nejběžnější metoda pro rozsah 3 kategorie 1 (zakoupené zboží a služby).
Přesnost založená na útratě vs
Metody založené na výdajích vytvářejí odhady s nejistotou 30–100 %, ve srovnání s 5-15 % metod založených na aktivitě. Používejte je pouze tehdy, když nemáte žádná data primárky. Protokol o emisích skleníkových plynů je přijímá jako výchozí bod, ale vyžaduje další postup zlepšení směrem k údajům o činnosti.
# scope3_spend_based.py
from typing import Optional
import httpx
# Mappa categorie NACE su activity_id Climatiq per spend-based
CATEGORY_TO_ACTIVITY_ID = {
# Settore IT e servizi digitali
"it_services": "professional_services-type_professional_services",
"cloud_hosting": "professional_services-type_professional_services",
"software_licenses": "professional_services-type_professional_services",
# Logistica e trasporti
"freight_road": "transport-type_freight_vehicle",
"freight_air": "transport-type_air_freight",
"courier_services": "transport-type_freight_vehicle",
# Produzione e manifattura
"raw_materials_steel": "steel-type_steel_products",
"raw_materials_plastic": "plastics-type_plastic_products",
"packaging": "paper-type_paper_products",
# Servizi professionali
"legal_services": "professional_services-type_professional_services",
"consulting": "professional_services-type_professional_services",
"marketing": "professional_services-type_professional_services",
# Utilities e energia
"electricity_bill": "electricity-supply_grid-source_residual_mix",
"gas_bill": "fuel_type-natural_gas",
}
async def calculate_scope3_spend_based(
api_key: str,
purchases: list[dict],
region: str = "IT",
currency: str = "EUR",
data_version: str = "^21"
) -> dict:
"""
Calcola Scope 3.1 con metodo spend-based.
purchases: [{"category": "it_services", "amount": 50000}, ...]
"""
BASE_URL = "https://beta3.api.climatiq.io"
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
batch_requests = []
for purchase in purchases:
activity_id = CATEGORY_TO_ACTIVITY_ID.get(purchase["category"])
if not activity_id:
print(f"Categoria non mappata: {purchase['category']}")
continue
batch_requests.append({
"emission_factor": {
"activity_id": activity_id,
"data_version": data_version,
"region": region
},
"parameters": {
"money": purchase["amount"],
"money_unit": currency.lower()
}
})
# Chunking per batch limite 100
results_all = []
chunk_size = 100
async with httpx.AsyncClient(timeout=60.0) as client:
for i in range(0, len(batch_requests), chunk_size):
chunk = batch_requests[i:i + chunk_size]
response = await client.post(
f"{BASE_URL}/batch/estimate",
headers=headers,
json={"requests": chunk}
)
response.raise_for_status()
results_all.extend(response.json()["results"])
# Aggrega per categoria
total_co2e_kg = 0.0
category_breakdown = {}
for purchase, result in zip(purchases, results_all):
if "error" in result:
category_breakdown[purchase["category"]] = {
"error": result["error"],
"amount": purchase["amount"]
}
continue
co2e_kg = result.get("co2e", 0.0)
total_co2e_kg += co2e_kg
cat = purchase["category"]
if cat not in category_breakdown:
category_breakdown[cat] = {
"total_co2e_kg": 0.0,
"total_spend": 0.0,
"intensity_kg_per_eur": 0.0
}
category_breakdown[cat]["total_co2e_kg"] += co2e_kg
category_breakdown[cat]["total_spend"] += purchase["amount"]
# Calcola intensità per categoria
for cat_data in category_breakdown.values():
if "total_spend" in cat_data and cat_data["total_spend"] > 0:
cat_data["intensity_kg_per_eur"] = (
cat_data["total_co2e_kg"] / cat_data["total_spend"]
)
return {
"scope": "scope_3.1",
"method": "spend_based",
"total_co2e_kg": total_co2e_kg,
"total_co2e_tco2e": total_co2e_kg / 1000,
"currency": currency,
"categories": category_breakdown,
"uncertainty_note": "Incertezza stimata 30-100% (metodo spend-based)"
}6. Integrace protokolu GHG: Rozsah mapování 1, 2, 3
Vybudování kompletního systému uhlíkového účetnictví vyžaduje mapování každého Kategorie protokolu GHG ke konkrétním koncovým bodům a Climatiq activity_id. Tato sekce poskytuje kompletní mapování pro nejrelevantnější kategorie.
| Rozsah protokolu GHG | Kategorie | Koncový bod Climatiq | Metoda |
|---|---|---|---|
| Rozsah 1 | Stacionární spalování (topení) | /estimate |
Aktivita (objem) |
| Rozsah 1 | Mobilní spalování (park) | /batch/estimate |
Aktivita (objem) |
| Rozsah 1 | Fugitivní emise (chladiva) | /estimate |
Aktivita (váha) |
| Rozsah 2 | Elektřina (podle polohy) | /energy |
Aktivita (kWh) |
| Rozsah 2 | Elektřina (tržní) | /energy |
Aktivita (kWh, zbytková směs) |
| Rozsah 3.1 | Nakoupené zboží a služby | /procurement |
Na základě útraty (EUR) |
| Rozsah 3.4 | Doprava proti proudu | /freight |
Činnost (tun-km) |
| Rozsah 3.6 | služební cesty (letadla) | /travel/flights |
Aktivita (kód IATA) |
| Rozsah 3.6 | Služební cesta (auto/vlak) | /estimate |
Aktivita (km) |
| Rozsah 3.7 | Dojíždění zaměstnanců | /batch/estimate |
Aktivita (km, polovina) |
| Rozsah 3.11 | Použití prodávaných produktů | /estimate |
Aktivita/útrata |
# ghg_protocol_calculator.py
# Sistema unificato per calcolo GHG Protocol completo
from dataclasses import dataclass, field
from typing import Optional
import asyncio
import httpx
@dataclass
class GHGProtocolReport:
"""Report GHG Protocol completo per anno fiscale."""
year: int
company_name: str
reporting_boundary: str = "operational_control"
# Scope 1
scope1_combustion_kg: float = 0.0
scope1_mobile_kg: float = 0.0
scope1_fugitive_kg: float = 0.0
# Scope 2
scope2_location_based_kg: float = 0.0
scope2_market_based_kg: float = 0.0
# Scope 3 (categorie principali)
scope3_cat1_purchased_goods_kg: float = 0.0
scope3_cat4_upstream_transport_kg: float = 0.0
scope3_cat6_business_travel_kg: float = 0.0
scope3_cat7_employee_commuting_kg: float = 0.0
# Metadata per audit trail
calculation_date: Optional[str] = None
data_version: str = ""
sources: list[str] = field(default_factory=list)
@property
def scope1_total_kg(self) -> float:
return (self.scope1_combustion_kg +
self.scope1_mobile_kg +
self.scope1_fugitive_kg)
@property
def scope2_selected_kg(self) -> float:
"""Market-based se disponibile, altrimenti location-based."""
return (self.scope2_market_based_kg
if self.scope2_market_based_kg > 0
else self.scope2_location_based_kg)
@property
def scope3_total_kg(self) -> float:
return (self.scope3_cat1_purchased_goods_kg +
self.scope3_cat4_upstream_transport_kg +
self.scope3_cat6_business_travel_kg +
self.scope3_cat7_employee_commuting_kg)
@property
def grand_total_tco2e(self) -> float:
return (self.scope1_total_kg +
self.scope2_selected_kg +
self.scope3_total_kg) / 1000
def to_csrd_dict(self) -> dict:
"""Output formato CSRD/ESRS E1-6."""
return {
"reporting_year": self.year,
"entity": self.company_name,
"ghg_emissions_location_based": {
"scope_1": self.scope1_total_kg / 1000,
"scope_2_location": self.scope2_location_based_kg / 1000,
"scope_3": self.scope3_total_kg / 1000,
"unit": "tCO2e"
},
"ghg_emissions_market_based": {
"scope_1": self.scope1_total_kg / 1000,
"scope_2_market": self.scope2_market_based_kg / 1000,
"scope_3": self.scope3_total_kg / 1000,
"unit": "tCO2e"
},
"data_version": self.data_version,
"methodology": "GHG Protocol Corporate Standard"
}7. Robustní klient Python: Opakovat pokus, vyrovnávací paměť a zpracování chyb
V produkci musí být volání API odolné vůči přechodným chybám, respekt omezení rychlosti a minimalizace hovorů pomocí efektivního ukládání do mezipaměti. Zde je klient připravený na výrobu.
# climatiq_client.py
import asyncio
import hashlib
import json
import logging
import time
from dataclasses import dataclass
from typing import Any, Optional
import httpx
import redis.asyncio as redis
logger = logging.getLogger(__name__)
class ClimatiqAPIError(Exception):
"""Errore API Climatiq con context."""
def __init__(self, message: str, status_code: int, response_body: dict):
super().__init__(message)
self.status_code = status_code
self.response_body = response_body
class RateLimitError(ClimatiqAPIError):
"""Rate limit superato - aspetta prima di riprovare."""
pass
class ClimatiqClient:
"""
Client asincrono per Climatiq API con:
- Retry automatico con exponential backoff
- Cache Redis per ridurre le chiamate API
- Logging strutturato per audit trail
- Gestione rate limit con rispetto dei retry-after header
"""
BASE_URL = "https://beta3.api.climatiq.io"
MAX_RETRIES = 3
BATCH_SIZE = 100 # Limite Climatiq
def __init__(
self,
api_key: str,
redis_client: Optional[redis.Redis] = None,
cache_ttl: int = 86400, # 24 ore - i fattori cambiano raramente
data_version: str = "^21"
):
self.api_key = api_key
self.redis = redis_client
self.cache_ttl = cache_ttl
self.data_version = data_version
self._http_client: Optional[httpx.AsyncClient] = None
async def __aenter__(self):
self._http_client = httpx.AsyncClient(
timeout=httpx.Timeout(30.0, connect=10.0),
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self._http_client:
await self._http_client.aclose()
def _cache_key(self, payload: dict) -> str:
"""Genera chiave cache deterministica da payload."""
payload_str = json.dumps(payload, sort_keys=True)
return f"climatiq:{hashlib.sha256(payload_str.encode()).hexdigest()[:16]}"
async def _get_cached(self, key: str) -> Optional[dict]:
"""Recupera risultato dalla cache Redis."""
if not self.redis:
return None
try:
cached = await self.redis.get(key)
if cached:
logger.debug(f"Cache HIT: {key}")
return json.loads(cached)
except Exception as e:
logger.warning(f"Errore cache GET: {e}")
return None
async def _set_cached(self, key: str, value: dict) -> None:
"""Salva risultato nella cache Redis."""
if not self.redis:
return
try:
await self.redis.setex(key, self.cache_ttl, json.dumps(value))
logger.debug(f"Cache SET: {key} (TTL: {self.cache_ttl}s)")
except Exception as e:
logger.warning(f"Errore cache SET: {e}")
async def _request_with_retry(
self, method: str, endpoint: str, payload: dict
) -> dict:
"""HTTP request con retry e exponential backoff."""
url = f"{self.BASE_URL}{endpoint}"
last_error = None
for attempt in range(self.MAX_RETRIES):
try:
response = await self._http_client.request(
method, url, json=payload
)
if response.status_code == 429: # Rate limit
retry_after = int(response.headers.get("Retry-After", 60))
logger.warning(
f"Rate limit superato. Aspetto {retry_after}s..."
)
await asyncio.sleep(retry_after)
continue
if response.status_code >= 400:
body = response.json() if response.content else {}
raise ClimatiqAPIError(
f"Errore API {response.status_code}: {body.get('error', 'Unknown')}",
status_code=response.status_code,
response_body=body
)
return response.json()
except httpx.NetworkError as e:
wait_time = 2 ** attempt
logger.warning(
f"Network error (attempt {attempt+1}/{self.MAX_RETRIES}), "
f"retry in {wait_time}s: {e}"
)
last_error = e
await asyncio.sleep(wait_time)
raise ConnectionError(
f"Falliti {self.MAX_RETRIES} tentativi: {last_error}"
)
async def estimate(
self,
activity_id: str,
parameters: dict,
region: Optional[str] = None
) -> dict:
"""
Stima singola con cache automatica.
activity_id: es. "electricity-supply_grid-source_residual_mix"
parameters: es. {"energy": 1000, "energy_unit": "kWh"}
"""
payload = {
"emission_factor": {
"activity_id": activity_id,
"data_version": self.data_version,
**({"region": region} if region else {})
},
"parameters": parameters
}
cache_key = self._cache_key(payload)
if cached := await self._get_cached(cache_key):
return cached
result = await self._request_with_retry("POST", "/estimate", payload)
await self._set_cached(cache_key, result)
logger.info(
"Estimate: activity=%(activity)s region=%(region)s "
"co2e=%(co2e).4f kg",
{
"activity": activity_id,
"region": region or "global",
"co2e": result.get("co2e", 0)
}
)
return result
async def batch_estimate(
self, requests: list[dict]
) -> list[dict]:
"""
Batch estimation con chunking automatico (100 per batch max).
"""
all_results = []
for i in range(0, len(requests), self.BATCH_SIZE):
chunk = requests[i:i + self.BATCH_SIZE]
payload = {"requests": chunk}
response = await self._request_with_retry(
"POST", "/batch/estimate", payload
)
all_results.extend(response.get("results", []))
# Piccola pausa tra chunk grandi per rispettare rate limit
if len(requests) > self.BATCH_SIZE:
await asyncio.sleep(0.5)
return all_results8. Klient TypeScript/Node.js s Axios
Pro backendové aplikace Node.js, TypeScript nebo NestJS je zde typovaný klient který odhaluje stejnou funkčnost jako klient Python s plnou typovou bezpečností.
// climatiq-client.ts
import axios, { AxiosInstance, AxiosError } from 'axios';
// Types per la Climatiq API
export interface EmissionFactor {
activity_id: string;
data_version: string;
region?: string;
}
export interface EstimateParameters {
energy?: number;
energy_unit?: 'kWh' | 'MWh' | 'GJ';
volume?: number;
volume_unit?: 'l' | 'gallon' | 'm3';
weight?: number;
weight_unit?: 'kg' | 't' | 'lb';
money?: number;
money_unit?: 'eur' | 'usd' | 'gbp';
distance?: number;
distance_unit?: 'km' | 'mi';
}
export interface EstimateRequest {
emission_factor: EmissionFactor;
parameters: EstimateParameters;
}
export interface ConstituentGases {
co2e_total: number;
co2?: number;
ch4?: number;
n2o?: number;
}
export interface EstimateResponse {
co2e: number;
co2e_unit: string;
co2e_calculation_method: string;
emission_factor: {
activity_id: string;
source: string;
year: number;
region: string;
category: string;
data_quality_flags: string[];
};
constituent_gases: ConstituentGases;
}
export interface BatchEstimateResult {
co2e?: number;
co2e_unit?: string;
error?: string;
constituent_gases?: ConstituentGases;
}
export class ClimatiqAPIError extends Error {
constructor(
message: string,
public readonly statusCode: number,
public readonly responseBody: unknown
) {
super(message);
this.name = 'ClimatiqAPIError';
}
}
export class ClimatiqClient {
private readonly http: AxiosInstance;
private readonly cache = new Map<string, { data: unknown; expiresAt: number }>();
private readonly cacheTtlMs: number;
constructor(
private readonly apiKey: string,
private readonly dataVersion = '^21',
cacheTtlSeconds = 3600
) {
this.cacheTtlMs = cacheTtlSeconds * 1000;
this.http = axios.create({
baseURL: 'https://beta3.api.climatiq.io',
headers: {
Authorization: `Bearer ${apiKey}`,
'Content-Type': 'application/json',
},
timeout: 30_000,
});
}
private cacheKey(payload: unknown): string {
return JSON.stringify(payload);
}
private getFromCache<T>(key: string): T | null {
const entry = this.cache.get(key);
if (!entry) return null;
if (Date.now() > entry.expiresAt) {
this.cache.delete(key);
return null;
}
return entry.data as T;
}
private setCache(key: string, data: unknown): void {
this.cache.set(key, {
data,
expiresAt: Date.now() + this.cacheTtlMs,
});
}
private handleAxiosError(error: AxiosError): never {
if (error.response) {
const status = error.response.status;
const body = error.response.data as Record<string, unknown>;
if (status === 429) {
throw new ClimatiqAPIError(
'Rate limit superato. Riprova tra qualche momento.',
429,
body
);
}
throw new ClimatiqAPIError(
`Climatiq API error ${status}: ${body['error'] ?? 'Unknown'}`,
status,
body
);
}
throw new ClimatiqAPIError(
`Network error: ${error.message}`,
0,
null
);
}
async estimate(
activityId: string,
parameters: EstimateParameters,
region?: string
): Promise<EstimateResponse> {
const payload: EstimateRequest = {
emission_factor: {
activity_id: activityId,
data_version: this.dataVersion,
...(region ? { region } : {}),
},
parameters,
};
const key = this.cacheKey(payload);
const cached = this.getFromCache<EstimateResponse>(key);
if (cached) return cached;
try {
const { data } = await this.http.post<EstimateResponse>(
'/estimate',
payload
);
this.setCache(key, data);
return data;
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
}
async batchEstimate(
requests: EstimateRequest[]
): Promise<BatchEstimateResult[]> {
const CHUNK_SIZE = 100;
const allResults: BatchEstimateResult[] = [];
for (let i = 0; i < requests.length; i += CHUNK_SIZE) {
const chunk = requests.slice(i, i + CHUNK_SIZE);
try {
const { data } = await this.http.post<{ results: BatchEstimateResult[] }>(
'/batch/estimate',
{ requests: chunk }
);
allResults.push(...data.results);
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
// Throttle tra chunk multipli
if (i + CHUNK_SIZE < requests.length) {
await new Promise(resolve => setTimeout(resolve, 200));
}
}
return allResults;
}
async estimateFlight(
originIata: string,
destinationIata: string,
passengers: number,
cabinClass: 'economy' | 'business' | 'first' = 'economy'
): Promise<EstimateResponse> {
try {
const { data } = await this.http.post<EstimateResponse>(
'/travel/flights',
{
legs: [{
from: originIata,
to: destinationIata,
passengers,
cabin_class: cabinClass,
}],
}
);
return data;
} catch (error) {
if (axios.isAxiosError(error)) this.handleAxiosError(error);
throw error;
}
}
}Použití klienta TypeScript
// usage-example.ts
import { ClimatiqClient } from './climatiq-client';
const client = new ClimatiqClient(
process.env['CLIMATIQ_API_KEY'] ?? '',
'^21',
3600
);
// Calcolo emissioni elettricità ufficio
async function calculateOfficeElectricity() {
const result = await client.estimate(
'electricity-supply_grid-source_residual_mix',
{ energy: 5000, energy_unit: 'kWh' },
'IT'
);
console.log(`Emissioni ufficio: ${result.co2e.toFixed(2)} kg CO2e`);
console.log(`Fonte: ${result.emission_factor.source} (${result.emission_factor.year})`);
return result;
}
// Calcolo batch per fleet management
async function calculateFleetEmissions(
vehicles: Array<{ id: string; litres: number; fuel: string }>
) {
const requests = vehicles.map(v => ({
emission_factor: {
activity_id: `fuel_type-${v.fuel}`,
data_version: '^21',
region: 'IT',
},
parameters: {
volume: v.litres,
volume_unit: 'l' as const,
},
}));
const results = await client.batchEstimate(requests);
return vehicles.map((v, i) => ({
vehicleId: v.id,
co2eKg: results[i].co2e ?? 0,
error: results[i].error,
}));
}
// Calcolo volo business travel
async function calculateBusinessFlight() {
const result = await client.estimateFlight('MXP', 'LHR', 2, 'economy');
console.log(`Volo MXP-LHR (2 pax): ${result.co2e.toFixed(1)} kg CO2e`);
}9. Real-Time Carbon Calculator pro SaaS
Silným případem použití pro Climatiq je přidat a uhlíkový štítek na e-commerce nebo SaaS produkty: ukažte uživateli dopad uhlíku odhad objednávky nebo akce před jejím potvrzením. To zvyšuje transparentnost a podporuje informovanou volbu spotřebitelů.
Architektura: Carbon Label API pro elektronický obchod
# carbon_label_api.py
# FastAPI endpoint per carbon label real-time su prodotti e-commerce
from fastapi import FastAPI, HTTPException, Depends, BackgroundTasks
from pydantic import BaseModel, Field
from typing import Optional
import asyncio
app = FastAPI(title="Carbon Label API", version="1.0.0")
class ProductCarbonRequest(BaseModel):
"""Richiesta calcolo carbon label per carrello."""
items: list[dict] = Field(
description="Lista prodotti con categoria e peso",
example=[
{"product_id": "SKU-123", "category": "electronics", "weight_kg": 0.5, "quantity": 1},
{"product_id": "SKU-456", "category": "clothing", "weight_kg": 0.3, "quantity": 2}
]
)
shipping_method: str = Field(
default="road",
description="Metodo spedizione: road, air, sea"
)
destination_region: str = Field(
default="IT",
description="Regione destinazione ISO 3166-2"
)
origin_region: str = Field(
default="CN",
description="Regione origine/produzione"
)
class CarbonLabelResponse(BaseModel):
"""Risposta con carbon label completa."""
total_co2e_kg: float
breakdown: dict
label: str # "A", "B", "C", "D", "E" come etichetta energetica EU
label_color: str
offset_cost_eur: float # Stima costo compensazione
trees_equivalent: float # Equivalente alberi annui
km_car_equivalent: float # Equivalente km auto
# Mappatura categoria prodotto su activity_id per produzione
PRODUCT_CATEGORY_PRODUCTION = {
"electronics": "electrical_equipment-type_small_electronics",
"clothing": "textiles-type_clothing",
"food": "food-type_mixed",
"furniture": "furniture-type_mixed",
"books": "paper-type_books",
"plastics": "plastics-type_plastic_products",
}
# Mappatura metodo spedizione su activity_id
SHIPPING_ACTIVITY_ID = {
"road": "transport-type_freight_vehicle-fuel_source_diesel-vehicle_type_hgv",
"air": "transport-type_air_freight",
"sea": "transport-type_sea_freight-route_type_container_ship",
}
def calculate_carbon_label(co2e_kg: float) -> tuple[str, str]:
"""
Calcola etichetta A-E basata su impatto carbonico.
Soglie ispirate alla proposta EU eco-label per e-commerce.
"""
if co2e_kg < 0.5:
return "A", "#2ecc71" # Verde - impatto molto basso
elif co2e_kg < 1.5:
return "B", "#27ae60" # Verde scuro - impatto basso
elif co2e_kg < 5.0:
return "C", "#f39c12" # Arancione - impatto medio
elif co2e_kg < 15.0:
return "D", "#e67e22" # Arancione scuro - impatto alto
else:
return "E", "#e74c3c" # Rosso - impatto molto alto
@app.post("/api/v1/carbon-label", response_model=CarbonLabelResponse)
async def get_carbon_label(
request: ProductCarbonRequest,
climatiq: ClimatiqClient = Depends(get_climatiq_client)
):
"""
Calcola carbon label real-time per un carrello e-commerce.
Considera produzione + imballaggio + spedizione.
"""
batch_requests = []
# 1. Emissioni produzione per ciascun prodotto
for item in request.items:
activity_id = PRODUCT_CATEGORY_PRODUCTION.get(
item["category"],
"manufactured_goods-type_mixed" # Fallback generico
)
# Calcolo per peso totale (qty * peso)
total_weight = item["weight_kg"] * item.get("quantity", 1)
batch_requests.append({
"emission_factor": {
"activity_id": activity_id,
"data_version": "^21",
"region": request.origin_region
},
"parameters": {
"weight": total_weight,
"weight_unit": "kg"
}
})
# 2. Emissioni spedizione (distanza stimata)
total_weight_kg = sum(
i["weight_kg"] * i.get("quantity", 1)
for i in request.items
)
# Stima distanza basata su regioni (semplificata)
estimated_distance_km = _estimate_distance(
request.origin_region,
request.destination_region
)
shipping_activity = SHIPPING_ACTIVITY_ID.get(
request.shipping_method,
SHIPPING_ACTIVITY_ID["road"]
)
# Freight: tonne * km = tonne-km
tonne_km = (total_weight_kg / 1000) * estimated_distance_km
batch_requests.append({
"emission_factor": {
"activity_id": shipping_activity,
"data_version": "^21"
},
"parameters": {
"weight": total_weight_kg,
"weight_unit": "kg",
"distance": estimated_distance_km,
"distance_unit": "km"
}
})
# Chiamata batch a Climatiq
try:
results = await climatiq.batch_estimate(batch_requests)
except ClimatiqAPIError as e:
raise HTTPException(
status_code=502,
detail=f"Errore calcolo emissioni: {str(e)}"
)
# Aggrega risultati
production_co2e = sum(
r.get("co2e", 0) for r in results[:-1] # Tutti tranne ultimo (spedizione)
)
shipping_co2e = results[-1].get("co2e", 0) if results else 0
total_co2e = production_co2e + shipping_co2e
label, color = calculate_carbon_label(total_co2e)
# Calcola equivalenze intuitive per l'utente
# Offset market rate ~15 EUR/tCO2e (mercato volontario 2025)
offset_cost = (total_co2e / 1000) * 15.0
# Un albero assorbe ~22 kg CO2/anno
trees_equivalent = total_co2e / 22.0
# Auto media emette ~0.21 kg CO2/km
km_car = total_co2e / 0.21
return CarbonLabelResponse(
total_co2e_kg=round(total_co2e, 3),
breakdown={
"production_kg": round(production_co2e, 3),
"shipping_kg": round(shipping_co2e, 3),
"shipping_method": request.shipping_method,
"distance_km": estimated_distance_km
},
label=label,
label_color=color,
offset_cost_eur=round(offset_cost, 2),
trees_equivalent=round(trees_equivalent, 1),
km_car_equivalent=round(km_car, 1)
)
def _estimate_distance(origin: str, destination: str) -> float:
"""Stima distanza tra regioni in km (lookup semplificato)."""
DISTANCES = {
("CN", "IT"): 9_000,
("DE", "IT"): 950,
("IT", "IT"): 300,
("US", "IT"): 8_500,
("IN", "IT"): 7_200,
}
key = (origin[:2].upper(), destination[:2].upper())
return DISTANCES.get(key, 5_000) # Default 5000km se sconosciutoFrontend Widget pro zobrazení uhlíkového štítku
// carbon-label.component.ts (Angular/TypeScript)
import { Component, Input, OnChanges, SimpleChanges } from '@angular/core';
import { HttpClient } from '@angular/common/http';
interface CarbonLabel {
total_co2e_kg: number;
label: string;
label_color: string;
trees_equivalent: number;
km_car_equivalent: number;
offset_cost_eur: number;
}
@Component({
selector: 'app-carbon-label',
template: `
<div class="carbon-label" *ngIf="carbonData">
<div class="label-badge" [style.background]="carbonData.label_color">
{{ carbonData.label }}
</div>
<div class="carbon-info">
<span class="co2e">{{ carbonData.total_co2e_kg | number:'1.1-2' }} kg CO₂e</span>
<span class="equivalent">
= {{ carbonData.km_car_equivalent | number:'1.0-0' }} km in auto
</span>
<button (click)="offsetCarbon()" class="offset-btn">
Compensa per €{{ carbonData.offset_cost_eur | number:'1.2-2' }}
</button>
</div>
</div>
`
})
export class CarbonLabelComponent implements OnChanges {
@Input() cartItems: Array<{ product_id: string; category: string; weight_kg: number; quantity: number }> = [];
@Input() shippingMethod = 'road';
carbonData: CarbonLabel | null = null;
loading = false;
constructor(private http: HttpClient) {}
ngOnChanges(changes: SimpleChanges): void {
if (changes['cartItems'] || changes['shippingMethod']) {
this.loadCarbonLabel();
}
}
private loadCarbonLabel(): void {
if (!this.cartItems.length) return;
this.loading = true;
this.http.post<CarbonLabel>('/api/v1/carbon-label', {
items: this.cartItems,
shipping_method: this.shippingMethod,
destination_region: 'IT',
origin_region: 'CN',
}).subscribe({
next: (data) => {
this.carbonData = data;
this.loading = false;
},
error: (err) => {
console.error('Errore carbon label:', err);
this.loading = false;
},
});
}
offsetCarbon(): void {
// Integrazione con piattaforma di offsetting
window.open('https://example.com/offset', '_blank');
}
}10. Testování a Mock API
V CI/CD nechcete skutečná volání rozhraní Climatiq API. Zde je návod, jak jej strukturovat testy s robustními simulacemi, které simulují reakce na úspěch i selhání.
# tests/conftest.py
import pytest
from unittest.mock import AsyncMock, MagicMock
from app.climatiq_client import ClimatiqClient
@pytest.fixture
def mock_climatiq_client():
"""
Mock del client Climatiq per test unitari.
Simula risposte realistiche senza chiamate API reali.
"""
client = AsyncMock(spec=ClimatiqClient)
# Risposta standard per stima elettricità italiana
electricity_response = {
"co2e": 415.0,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"emission_factor": {
"activity_id": "electricity-supply_grid-source_residual_mix",
"source": "IEA",
"year": 2022,
"region": "IT",
"category": "Electricity",
"data_quality_flags": []
},
"constituent_gases": {
"co2e_total": 415.0,
"co2": 415.0,
"ch4": None,
"n2o": None
}
}
# Risposta per diesel
diesel_response = {
"co2e": 302.15,
"co2e_unit": "kg",
"co2e_calculation_method": "ar5",
"emission_factor": {
"activity_id": "fuel_type-diesel",
"source": "DEFRA",
"year": 2023,
"region": "IT",
"category": "Fuel"
},
"constituent_gases": {
"co2e_total": 302.15,
"co2": 301.5,
"ch4": 0.02,
"n2o": 0.63
}
}
# Configura mock per rispondere in base all'activity_id
async def mock_estimate(activity_id, parameters, region=None):
if "electricity" in activity_id:
kwh = parameters.get("energy", 1000)
return {**electricity_response, "co2e": kwh * 0.415}
elif "diesel" in activity_id:
litres = parameters.get("volume", 100)
return {**diesel_response, "co2e": litres * 2.52}
return electricity_response
client.estimate = AsyncMock(side_effect=mock_estimate)
client.batch_estimate = AsyncMock(return_value=[electricity_response])
return client
# tests/test_activity_based.py
import pytest
from unittest.mock import AsyncMock
@pytest.mark.asyncio
async def test_fleet_scope1_calculates_correctly(mock_climatiq_client):
"""Test che il calcolo Scope 1 flotta produca risultati corretti."""
from app.services.fleet_calculator import calculate_fleet_scope1
result = await calculate_fleet_scope1(
client=mock_climatiq_client,
litres_diesel=100.0,
region="IT"
)
assert result["co2e_kg"] == pytest.approx(252.0, rel=0.01)
assert result["scope"] == "scope_1"
mock_climatiq_client.estimate.assert_called_once()
@pytest.mark.asyncio
async def test_rate_limit_error_is_handled(mock_climatiq_client):
"""Test che rate limit error non lasci il sistema in stato inconsistente."""
from app.climatiq_client import ClimatiqAPIError
mock_climatiq_client.estimate = AsyncMock(
side_effect=ClimatiqAPIError("Rate limit", 429, {"error": "rate_limit"})
)
with pytest.raises(ClimatiqAPIError) as exc_info:
await mock_climatiq_client.estimate(
"electricity-supply_grid-source_residual_mix",
{"energy": 1000, "energy_unit": "kWh"},
"IT"
)
assert exc_info.value.status_code == 429
@pytest.mark.asyncio
async def test_scope2_100_renewable_returns_zero(mock_climatiq_client):
"""Test che 100% rinnovabile dia emissioni zero (market-based)."""
from app.services.scope2_calculator import (
calculate_scope2_electricity, Scope2Method
)
result = await calculate_scope2_electricity(
client=mock_climatiq_client,
kwh_consumed=100_000,
region="IT",
method=Scope2Method.MARKET_BASED,
renewable_percentage=100.0
)
assert result["co2e_kg"] == 0.0
# Non deve chiamare l'API (nessun consumo da rete)
mock_climatiq_client.estimate.assert_not_called()11. Odhad dávky a strategie ukládání do mezipaměti pro Scala
Pro aplikace, které zpracovávají tisíce záznamů (měsíční zprávy, historické analýzy), kombinace dávkových volání API a mezipaměti Redis se stává nezbytnou pro snížení jak doby zpracování, tak náklady na API.
# batch_processor.py
# Elaborazione batch per calcoli di emissioni su larga scala
import asyncio
from datetime import datetime
from typing import AsyncIterator
import redis.asyncio as redis
async def process_monthly_emissions_report(
api_key: str,
records: list[dict],
redis_url: str = "redis://localhost:6379"
) -> dict:
"""
Elabora report mensile emissioni per grandi dataset.
records: lista di attività (flotta, energia, acquisti, etc.)
Restituisce: aggregato mensile Scope 1, 2, 3.
"""
redis_client = await redis.from_url(redis_url)
async with ClimatiqClient(
api_key=api_key,
redis_client=redis_client,
cache_ttl=86400 * 7, # 7 giorni per fattori stabili
data_version="21" # Versione fissa per riproducibilità
) as client:
# Raggruppa per tipo di emissione
scope1_records = [r for r in records if r["scope"] == "scope1"]
scope2_records = [r for r in records if r["scope"] == "scope2"]
scope3_records = [r for r in records if r["scope"] == "scope3"]
# Elabora in parallelo i tre scope
scope1_result, scope2_result, scope3_result = await asyncio.gather(
_process_scope1_batch(client, scope1_records),
_process_scope2_batch(client, scope2_records),
_process_scope3_batch(client, scope3_records)
)
total = (
scope1_result["total_co2e_kg"] +
scope2_result["total_co2e_kg"] +
scope3_result["total_co2e_kg"]
)
await redis_client.aclose()
return {
"report_generated_at": datetime.utcnow().isoformat(),
"total_co2e_kg": total,
"total_co2e_tco2e": total / 1000,
"scope1": scope1_result,
"scope2": scope2_result,
"scope3": scope3_result,
"record_count": len(records)
}
async def _process_scope1_batch(
client: ClimatiqClient, records: list[dict]
) -> dict:
"""Elabora batch Scope 1 con chunking automatico."""
if not records:
return {"total_co2e_kg": 0.0, "record_count": 0}
batch_requests = [
{
"emission_factor": {
"activity_id": r["activity_id"],
"data_version": "21",
"region": r.get("region", "IT")
},
"parameters": {
r["param_key"]: r["param_value"],
f"{r['param_key']}_unit": r["param_unit"]
}
}
for r in records
]
results = await client.batch_estimate(batch_requests)
total = sum(r.get("co2e", 0) for r in results if "error" not in r)
errors = [r for r in results if "error" in r]
if errors:
import logging
logging.warning(f"{len(errors)} errori nel batch Scope 1")
return {
"total_co2e_kg": total,
"record_count": len(records),
"error_count": len(errors)
}12. Alternativy k Climatiq: Srovnání
Climatiq není jedinou možností. Zde je srovnání těch hlavních alternativy, které vám pomohou vybrat správné řešení:
| Řešení | Hlavní síla | Omezit | Ideální případ použití | Volný plán |
|---|---|---|---|---|
| Climatiq | 190 000+ faktorů, 40+ zdrojů, robustní API | 250 hovorů/měsíc v komunitním plánu | Podniková, víceoborová výroba | 250 hovorů/měsíc |
| Uhlíkové rozhraní | Jednoduché API, skvělé vývojářské zkušenosti | Méně faktorů, zaměřit se na USA | Startupy, e-commerce, doprava | Ano (omezeno) |
| Otevřené emisní faktory | Zdarma, open source, spravovaný Climatiq | Pouze datové sady, žádné přímé REST API | Výzkum, přístup k nezpracovaným datům | Ano (datová sada) |
| Rozhraní API | Podnikové funkce připravené na audit | Pouze pro podniky, drahé | Velké společnosti, reporting CSRD | No |
| Databáze EPA FLIGHT | Zdarma, vláda USA | Pouze USA, nikoli REST API | Americké zpravodajství, výzkum | Ano (datová sada) |
| Vlastní databáze (DEFRA/IEA) | Plná kontrola, žádné náklady na API | Nákladná vnitřní údržba | Velké organizace s oddaným týmem | Ano (veřejná data) |
Kdy použít vlastní databázi
V některých případech dává vybudování interní databáze emisních faktorů smysl. Zde je schéma PostgreSQL, jak to zvládnout:
-- Schema PostgreSQL per database fattori di emissione personalizzato
CREATE TABLE IF NOT EXISTS emission_factors (
id SERIAL PRIMARY KEY,
activity_category VARCHAR(200) NOT NULL,
activity_name VARCHAR(500),
region_code VARCHAR(10) NOT NULL,
reference_year INTEGER NOT NULL,
unit_type VARCHAR(50) NOT NULL, -- 'litre', 'kWh', 'tonne', 'EUR'
factor_kg_co2e_per_unit DECIMAL(12, 6) NOT NULL,
source VARCHAR(100) NOT NULL, -- 'DEFRA', 'ISPRA', 'EPA'
source_version VARCHAR(50),
source_priority INTEGER DEFAULT 1,
constituent_co2 DECIMAL(12, 6),
constituent_ch4 DECIMAL(12, 6),
constituent_n2o DECIMAL(12, 6),
created_at TIMESTAMP DEFAULT NOW(),
updated_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_ef_lookup ON emission_factors
(activity_category, region_code, reference_year, unit_type);
-- Query di lookup con fallback regionale
SELECT factor_kg_co2e_per_unit, source, reference_year
FROM emission_factors
WHERE activity_category = $1
AND reference_year = $2
AND unit_type = $3
AND region_code IN ($4, 'EU', 'GLOBAL') -- Fallback a EU poi globale
ORDER BY
CASE region_code
WHEN $4 THEN 1 -- Regione specifica prima
WHEN 'EU' THEN 2 -- Poi EU
ELSE 3 -- Poi globale
END,
source_priority DESC,
reference_year DESC
LIMIT 1;Když vlastní databáze NENÍ tou správnou volbou
Vytvoření a správa databáze emisních faktorů vyžaduje: oddanou osobou po dobu 2-3 měsíců pro počáteční stavbu, povinné pololetní aktualizace (faktory se každý rok mění), správa verzí pro audit trail a geografické pokrytí tak často chybí pro regiony mimo OECD. Pro většinu společností, Climatiq nabízí vynikající návratnost investic i s ohledem na cenu API.
13. Případová studie: Platforma elektronického obchodu s Carbon Label
EcoShop Itálie je platforma elektronického obchodování s 50 000 produkty a 200 000 objednávek/měsíc. Požadavek: přidat uhlíkový štítek na každý produkt a ukázat odhadované emise při pokladně s možností kompenzace.
Implementovaná architektura
- Rozšíření katalogu produktů: Noční práce, která předem kalkuluje produkční emise pro každou SKU pomocí koncového bodu šarže. Výsledky uloženy v databázi produktů s 30denní TTL.
- Pokladna v reálném čase: Na pokladně dynamický výpočet emisí doprava na základě hmotnosti košíku, cílové oblasti a zvoleného způsobu. Cílová latence: pod 200 ms (s mezipamětí Redis pro standardní zásilky).
- Prodejce karbonové palubní desky: Měsíční zpráva pro prodejce s Rozsah 3.4 (předchozí doprava) a odhadovaný rozsah 3.11 (použití prodávaných produktů).
# ecoshop_carbon_service.py
# Servizio di calcolo emissioni per EcoShop Italia
from dataclasses import dataclass
from typing import Optional
import asyncio
from datetime import datetime, timedelta
@dataclass
class ProductEmissionProfile:
"""Profilo emissioni di un prodotto nel catalogo."""
sku: str
production_co2e_kg: float
packaging_co2e_kg: float
category: str
origin_region: str
calculated_at: datetime
climatiq_data_version: str
@property
def total_product_co2e_kg(self) -> float:
return self.production_co2e_kg + self.packaging_co2e_kg
@property
def is_stale(self) -> bool:
"""Profilo è obsoleto se più vecchio di 30 giorni."""
return datetime.utcnow() - self.calculated_at > timedelta(days=30)
class EcoShopCarbonService:
"""
Servizio carbon per EcoShop: gestisce calcoli produzione,
spedizione e report venditori.
"""
def __init__(
self,
climatiq_client: ClimatiqClient,
db_pool, # asyncpg pool
):
self.climatiq = climatiq_client
self.db = db_pool
async def get_product_emission_profile(
self, sku: str
) -> Optional[ProductEmissionProfile]:
"""
Recupera profilo emissioni dal DB.
Se assente o obsoleto, ricalcola via Climatiq.
"""
# Controlla DB
profile = await self._load_from_db(sku)
if profile and not profile.is_stale:
return profile
# Ricalcola
product = await self._get_product_data(sku)
if not product:
return None
new_profile = await self._calculate_product_emissions(product)
await self._save_to_db(new_profile)
return new_profile
async def calculate_checkout_carbon(
self,
cart_items: list[dict],
destination_region: str,
shipping_method: str
) -> dict:
"""
Calcola carbon label per checkout in tempo reale.
Target: < 200ms con cache.
"""
# Recupera profili prodotti (con cache)
profiles = await asyncio.gather(*[
self.get_product_emission_profile(item["sku"])
for item in cart_items
])
# Calcola emissioni produzione
production_co2e = sum(
(profile.total_product_co2e_kg * item.get("quantity", 1))
for profile, item in zip(profiles, cart_items)
if profile is not None
)
# Calcola emissioni spedizione
total_weight_kg = sum(
item.get("weight_kg", 0.5) * item.get("quantity", 1)
for item in cart_items
)
shipping_co2e = await self._estimate_shipping(
total_weight_kg, destination_region, shipping_method
)
total = production_co2e + shipping_co2e
label, color = calculate_carbon_label(total)
return {
"total_co2e_kg": round(total, 3),
"production_co2e_kg": round(production_co2e, 3),
"shipping_co2e_kg": round(shipping_co2e, 3),
"label": label,
"label_color": color,
"offset_price_eur": round((total / 1000) * 15.0, 2),
"generated_at": datetime.utcnow().isoformat()
}
# Risultati dopo 6 mesi
ECOSHOP_METRICS = {
"prodotti_con_carbon_label": 47_832,
"ordini_con_label_mese": 180_000,
"percentuale_utenti_offset": 8.3, # % utenti che comprano offset
"revenue_offset_mensile_eur": 4_200,
"latenza_media_ms": 45, # ms (grazie a cache Redis)
"api_calls_risparmiate_cache": "92%",
"co2e_totale_calcolato_mese_tco2e": 1_240,
"nps_incremento": +12 # punti NPS grazie a trasparenza
}14. Nasazení a bezpečná konfigurace ve výrobě
# app/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
from functools import lru_cache
class Settings(BaseSettings):
"""Configurazione applicazione da variabili d'ambiente."""
# Climatiq
climatiq_api_key: str
climatiq_data_version: str = "^21"
# Cache
cache_ttl_seconds: int = 3600
redis_url: str = "redis://localhost:6379"
# API Security
api_secret_key: str
allowed_origins: list[str] = ["https://dashboard.tuaazienda.it"]
# Database (per audit trail)
database_url: str = "postgresql://user:pass@localhost/ghg_db"
model_config = SettingsConfigDict(env_file=".env", case_sensitive=False)
@lru_cache
def get_settings() -> Settings:
return Settings()
# .env (NON committare in git - usare secrets manager in produzione)
# CLIMATIQ_API_KEY=clq_live_xxxxxxxxxxxxxxxxxxxxxxx
# CLIMATIQ_DATA_VERSION=^21
# API_SECRET_KEY=generato-con-openssl-rand-hex-32
# CACHE_TTL_SECONDS=3600
# REDIS_URL=redis://redis:6379
# DATABASE_URL=postgresql://ghg_user:password@postgres:5432/ghg_dbNikdy nevystavujte klíč API Climatiq
Klíč Climatiq API nikdy nesmí být ve zdrojovém kódu, v protokolech, v odpovědích API nebo proměnných JavaScriptu na straně klienta. Vždy používejte: proměnné prostředí, správce tajemství (AWS Secrets Manager, HashiCorp Vault, Azure Key Vault) nebo Kubernetes Secrets s klidovým šifrováním. V případě náhodného odhalení klíčem ihned otočte.
15. Nejlepší postupy a anti-vzorce
Nejlepší postupy
-
Uzamknout data_version v produkci: USA
"21"místo"^21"aby byla zajištěna reprodukovatelnost výpočtů. Záměrně aktualizujte explicitním přepočtem. - Agresivní ukládání do mezipaměti: Emisní faktory se mění jen zřídka. 24-168hodinová mezipaměť dramaticky snižuje volání API, např pro opakované výpočty (měsíční vozový park, produktový katalog).
- Vždy auditní záznam: Uložte každý výpočet pomocí activity_id, použitý faktor, verze dat, časové razítko a zdroj. Nezbytné pro CSRD.
- Pokud je to možné, použijte dávku: Jediný hromadný hovor s 100 položek je mnohem efektivnější než 100 jednotlivých hovorů. Respektuje rychlostní limity a snižuje celkovou latenci.
- Ověřit data_quality_flags: Climatiq hlásí, když a faktor má nízkou kvalitu dat. Řešte tyto případy pomocí alternativních faktorů nebo poznámky o nejistotě ve zprávě.
- Duální hlášení pro rozsah 2: Protokol GHG vyžaduje jak lokační, tak tržní metody. Spočítejte a uložte obojí.
- Mezipaměť v paměti pro TypeScript: Implementujte mapu s TTL v klientovi TypeScript, abyste se vyhnuli nadbytečným voláním v aplikacích Dlouhotrvající Node.js bez Redis.
Anti-vzory, kterým je třeba se vyhnout
- Klíč API ve frontendu: Nevystavujte klíč Climatiq JavaScript na straně klienta. Vždy to jde přes backend.
- Ignorujte složkové plyny: Pro přesné hlášení CSRD, CO₂, CH4 a N2O musí být vykazovány odděleně kromě celkového CO₂e.
- Nesprávné jednotky měření: Climatiq přijímá převody mezi jednotky, ale procházející litry, když koncový bod očekává výsledky v kWh numericky věrohodné, ale vědecky špatné. Vždy platné.
- Dávka > 100 položek: API vrací chybu 422. Vždy implementujte logiku chunkingu pro velké datové sady.
-
Ignorovat shodu regionů: Faktor pro elektřinu
bez určení oblasti použijte globální výchozí nastavení. Pro Itálii vždy používejte
"IT". -
Blokování synchronních hovorů: Nikdy nepoužívejte synchronní HTTP knihovny
v asynchronních koncových bodech. Vždy používejte
httpx.AsyncClientv Pythonu neboaxiosconasync/awaitv TypeScriptu.
Závěry a další kroky
Climatiq řeší nejobtížnější problém automatického účtování uhlíku: a databáze emisních faktorů. S více než 190 000 ověřenými faktory pokrytí 300 regionů a průběžné aktualizace, umožňuje stavět systémy pro výpočet emisí skleníkových plynů připravené k výrobě ve dnech namísto měsíců.
V tomto článku jsme vytvořili:
- Un Robustní Python klient s opakováním, mezipamětí Redis a zpracováním zadaných chyb
- Un Klient TypeScript/Node.js s Axios a plnou bezpečností pro frontend integrace
- Výpočty pro Rozsah 1 (dieselový/HVO vozový park), Rozsah 2 (dvojí metoda) a hlavní kategorie Rozsah 3
- Un Carbon Label API v reálném čase pro elektronický obchod s A-E štítkem a offsety
- Testování s falešné API pro prostředí CI/CD bez skutečných hovorů
- Strategie dávkové zpracování a ukládání do mezipaměti pro podnikové měřítko
Green Software Series pokračuje
- Předchozí článek: CodeCarbon - Měření emisí kódu běžící s open source knihovnou Python.
- Další článek: Carbon Aware SDK – Jak přesunout pracovní zátěž v hodinách s nejčistší energií pomocí předpovědi intenzity sítě.
- Související článek (série MLOps): Optimalizovat školení modelů ML ke snížení emisí uhlíku.
- Související článek (Data & AI Business Series): Správa dat for Reliable AI – Jak integrovat metriky udržitelnosti do katalogu dat.
Dalším praktickým krokem je přihlášení do komunitního plánu společnosti Climatiq (250 bezplatných volání měsíčně), prozkoumejte Průzkumník dat najít faktory relevantní pro vaše odvětví a implementovat je výpočet rozsahu 1 pro váš nejjednodušší případ použití jako první milník.
S regulačními tlaky CSRD (povinné pro velké společnosti v EU od roku 2025, rozšířena na malé a střední podniky od roku 2026) a rostoucí pozornost investorů k metrikám ESG, mít technickou infrastrukturu pro automatizované uhlíkové účtování již není konkurenční výhoda: je to a provozní nutnost.