Climatiq API: Integrați calculele emisiilor de GES în backend
În 2025, piața soluțiilor de monitorizare a emisiilor de carbon a crescut până când 17,3 miliarde de dolari, condusă de reglementări din ce în ce mai stricte precum CSRD european, divulgarea SEC privind clima în SUA și ISO 14064. Companiile nu pot nu te mai baza pe foile Excel și estimările brute: sunt utile calcule automate, verificabil și integrat în sistemele de operare.
Provocarea tehnică este reală: calcularea emisiilor de GES (gaze cu efect de seră) necesită acces la baze de date factori de emisie actualizate, metodologii validate prin Protocolul GHG și conversii dintre sute de unități de măsură diferite. Construirea și întreținerea tuturor acestor lucruri în interior necesită luni de muncă specializată. Aici intervine Climatiq.
Climatiq este un API REST care oferă acces la over 190.000 de factori de emisie din peste 40 de surse verificate (EPA, DEFRA/BEIS, IEA, ecoinvent), acoperirea Peste 300 de regiuni geografice, și calcule conforme cu Protocolul GHG pentru Scopul 1, 2 și 3. În acest articol vom construi un Backend complet FastAPI care integrează Climatiq pentru calculele emisiilor de producție, împreună cu un client TypeScript și un calculator în timp real pentru aplicații SaaS.
Ce vei învăța
- Arhitectura Climatiq API: puncte finale, autentificare, limite de rată și modele de date
- Baza de date privind factorii de emisie: cum să căutați, să filtrați și să selectați factorii corecti
- Estimare bazată pe activitate: calcule bazate pe activități concrete (kWh, km, kg)
- Estimare bazată pe cheltuieli: calcule ale cheltuielilor valutare atunci când lipsesc datele primare
- GHG Protocol Scope 1, 2, 3: cartografierea categoriei și calcule conforme
- Client Python robust cu reîncercare, memorare în cache Redis și tratare a erorilor
- Client TypeScript/Node.js cu Axios și tipuri pentru integrarea front-end
- Calculator de carbon în timp real pentru SaaS cu etichete de carbon pe produse
- Testare cu API simulată pentru medii CI/CD
- Alternative la Climatiq și comparație de funcționalități
Seria Green Software Engineering
Acest articol face parte din seria completă despre Green Software Engineering. Fiecare articol abordează un aspect specific al sustenabilității digitale:
| # | Articol | Subiect principal |
|---|---|---|
| 1 | Principii Green Software Engineering | GSF, SCI spec, 8 principii fundamentale |
| 2 | CodeCarbon: Măsurarea emisiilor de cod | Bibliotecă Python, tablou de bord, integrare CI/CD |
| 3 | Climatiq API: calcule GHG în backend | REST API, Scope 1-3, FastAPI + Integrare TypeScript |
| 4 | SDK Carbon Aware | Schimbarea sarcinii de lucru, intensitatea grilei, schimbarea timpului |
| 5 | Domeniul 1, 2 și 3: Modelarea datelor pentru raportarea ESG | Structura datelor, calcule, agregare, raportare |
| 6 | GreenOps: Infrastructură conștientă de carbon | Programare Kubernetes, scalare în funcție de probleme |
| 7 | Lanțul valoric Scope 3 al conductei de emisii | Colectarea datelor furnizorului, calcul, pista de audit |
| 8 | API de raportare ESG: Integrare CSRD | Flux de lucru CSRD, automatizare de rapoarte, conformitate |
| 9 | Modele arhitecturale durabile | Stocare, stocare în cache inteligentă, lot care ține seama de carbon |
| 10 | AI și carbon: amprenta antrenamentului ML | Emisii LLM, optimizare, Green AI |
1. Protocolul GHG și necesitatea calculelor automate
Il Standardul corporativ al protocolului GHG este cel mai adoptat cadru din lume pentru contabilizarea emisiilor companiei. Clasifică emisiile în trei domenii:
- Domeniul 1 (emisii directe): Arderea combustibililor în vehiculele companiei, instalatii de productie, incalzire. Sunt sub controlul direct al companiei.
- Domeniul 2 (Energie indirectă): Cumpărat energie electrică, abur, căldură. Ele sunt împărțite în bazate pe locație (mix de rețea locală) e bazate pe piata (certificate energetice, PPA).
- Domeniul 3 (lanțul valoric): 15 categorii care includ achiziționarea de bunuri și servicii, transport în amonte/aval, utilizare a produsului, sfârșit de viață, călătorii de afaceri, naveta angajaților și multe altele. Ele reprezintă de obicei 70-90% a emisiilor totale a unei companii.
Pentru fiecare calcul ai nevoie de un factor de emisie: un coeficient care convertește o activitate (litri de motorină, kWh de energie electrică, euro de cumpărare) în kg CO₂e. Acești factori variază după:
- Anul de referinta: rețelele de energie electrică schimbă mixul în fiecare an
- Regiunea geografică: kWh italian este diferit de kWh german
- Sursa datelor: EPA pentru SUA, DEFRA pentru Marea Britanie, ISPRA pentru Italia
- Perimetru: în amonte, în aval, de la leagăn la poartă, de la leagăn la mormânt
Menținerea unei baze de date actualizate cu factori de emisie necesită o echipă dedicată. Climatiq o face pentru noi, agregând în continuare 40 de surse verificate cu actualizări continue.
2. Prezentare generală a API-ului Climatiq
Arhitectură și puncte finale principale
API-ul Climatiq este un API REST JSON bazat pe URL https://beta3.api.climatiq.io.
Autentificarea se face prin Jeton de purtător în antetul HTTP.
Planurile disponibile includ:
| Podea | Apeluri/lună | Funcționalitate | Utilizare tipică |
|---|---|---|---|
| Comunitate | 250 | Toate punctele finale | Prototiparea, testarea |
| Starter | 5.000 | + Factori privați | IMM-uri, MVP |
| Creştere | 50.000 | + SLA, suport | Companii în creștere |
| Întreprindere | Personalizat | + Pista de audit, SSO | Organizații mari |
Obiectivele principale sunt:
| Puncte finale | Metodă | Descriere |
|---|---|---|
/search |
OBŢINE | Caută factori de emisie în baza de date |
/estimate |
POST | Emisia unică estimată din activități |
/batch/estimate |
POST | Estimări multiple (până la 100 per cerere) |
/travel/flights |
POST | Emisiile companiilor aeriene (domeniul de aplicare 3.6) |
/freight |
POST | Emisiile transportului de marfă multimodal |
/procurement |
POST | Probleme legate de achiziție (Scope 3.1, bazate pe cheltuieli) |
/energy |
POST | Emisii de consum de energie (Scope 2) |
/compute |
POST | Emisii de cloud computing |
Structura unei cereri de estimare
# 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
}
}Versiunea datelor: semantică și bune practici
Câmpul data_version controlează ce versiune a bazei de date să folosească.
Cartul (^21) utilizați versiunea 21 sau mai mare compatibilă, asigurând
actualizări automate ale factorilor de emisie. In productie, bloc
versiunea exactă (de ex. "21") pentru reproductibilitatea
calcule și piste de audit. Actualizați în mod deliberat la noi versiuni de software
cu recalcularea explicită a valorilor istorice.
3. Model de date: Baza de date privind factorii de emisie
Structura unui factor de emisie
Un factor de emisie din Climatiq este identificat în mod unic prin
activity_id, source, region, year
e lca_activity. Înțelegerea acestei structuri este critică
pentru a selecta factorul corect.
# 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
}
}Principalele surse de date
| Sursă | Țara/Regiune | Actualizare | Sectoarele acoperite |
|---|---|---|---|
| DEFRA/BEIS | UK | Anual | Energie, transport, achizitii, materiale |
| EPA | STATELE UNITE ALE AMERICII | Anual | Energie, procese industriale, agricultură |
| IEA | Global | Anual | Electricitate pe țară, energie primară |
| ecoinvent | Global | Semianual | LCA complet, lanț de aprovizionare, materiale |
| ADEME | Franţa | Anual | Pe bază de cărbune, transport, energie FR |
| SEE | EU | Anual | Rețea de energie electrică Europa, emisii sectoriale |
| ISPRA | Italia | Anual | Inventarul național de GES, mixul de energie electrică IT |
4. Estimare bazată pe activitate: calcule pe activități concrete
Abordarea bazate pe activitate este cea mai precisă: se bazează pe date de activitate primară, cum ar fi litri de combustibil consumat, kWh de energie electrică, kilometri parcurși sau tone de material achiziționat. Necesită colectarea datelor operaționale detaliat, dar produce rezultate solide din punct de vedere științific.
Exemplu: calculul emisiilor flotei companiei (domeniul de aplicare 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())Calculul emisiilor de energie electrică (Scope 2)
Scopul 2 necesită calcul cu două metode: bazat pe locație (mix de rețea) și bazate pe piață (factor de amestec rezidual sau PPA). Ambele valori trebuie raportate în divulgarea Protocolului 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. Estimare bazată pe cheltuieli: calcule din datele privind cheltuielile
Când nu ai date precise de activitate, abordarea bazat pe cheltuieli folosește cheltuielile valutare ca proxy pentru activitate. Climatiq aplică factori de emisie economic (kg CO₂e per euro/dolar cheltuit) pe categorie de achiziție, pe baza datelor input-output din tabelele OCDE sau EXIOBASE. Este cea mai comună metodă pentru Scope 3 categoria 1 (Bunuri și servicii achiziționate).
Precizie bazată pe cheltuieli vs. Precizie bazată pe activitate
Metodele bazate pe cheltuieli produc estimări cu incertitudine de 30-100%, comparativ cu 5-15% din metodele bazate pe activitate. Folosiți-le numai atunci când nu aveți date primare. Protocolul GHG le acceptă ca punct de plecare, dar necesită progresie îmbunătățirea datelor privind activitatea.
# 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. Integrarea protocolului GHG: Scopul de cartografiere 1, 2, 3
Construirea unui sistem complet de contabilizare a carbonului necesită cartografierea fiecăruia GHG Protocol categorie la anumite puncte finale și Climatiq activity_id. Această secțiune oferă cartografiere completă pentru cele mai relevante categorii.
| Domeniul de aplicare GHG Protocol | Categorie | Endpoint Climatiq | Metodă |
|---|---|---|---|
| Domeniul de aplicare 1 | Combustie staționară (încălzire) | /estimate |
Activitate (volum) |
| Domeniul de aplicare 1 | Combustie mobilă (flotă) | /batch/estimate |
Activitate (volum) |
| Domeniul de aplicare 1 | Emisii fugitive (agenți frigorifici) | /estimate |
Activitate (greutate) |
| Domeniul de aplicare 2 | Electricitate (în funcție de locație) | /energy |
Activitate (kWh) |
| Domeniul de aplicare 2 | Electricitate (pe piață) | /energy |
Activitate (kWh, amestec rezidual) |
| Domeniul de aplicare 3.1 | Bunuri și servicii achiziționate | /procurement |
Bazat pe cheltuieli (EUR) |
| Domeniul de aplicare 3.4 | Transport în amonte | /freight |
Activitate (tonă-km) |
| Domeniul de aplicare 3.6 | Călătorii de afaceri (avioane) | /travel/flights |
Activitate (cod IATA) |
| Domeniul de aplicare 3.6 | Călătorii de afaceri (mașină/tren) | /estimate |
Activitate (km) |
| Domeniul de aplicare 3.7 | Naveta angajaților | /batch/estimate |
Activitate (km, jumătate) |
| Domeniul de aplicare 3.11 | Utilizarea produselor comercializate | /estimate |
Activitate/cheltuieli |
# 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. Client Python robust: Reîncercați, cache și gestionarea erorilor
În producție, apelurile API trebuie să fie rezistente la erori tranzitorii, respect limitele de rată și minimizați apelurile cu memorarea în cache eficientă. Iată un client gata de producție.
# 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. Client TypeScript/Node.js cu Axios
Pentru aplicațiile backend Node.js, TypeScript sau NestJS, iată un client tastat care expune aceeași funcționalitate ca și clientul Python cu siguranță de tip complet.
// 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;
}
}
}Utilizarea clientului 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. Calculator de carbon în timp real pentru SaaS
Un caz de utilizare puternic pentru Climatiq este adăugarea unui eticheta de carbon către comerțul electronic sau produsele SaaS: arată utilizatorului impactul carbonului estimarea unei comenzi sau acțiuni înainte de a o confirma. Acest lucru crește transparența și sprijină alegerile informate ale consumatorilor.
Arhitectură: Carbon Label API pentru comerțul electronic
# 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 sconosciutoWidget de front-end pentru a afișa eticheta de carbon
// 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. Testare și simulare API
În CI/CD, nu doriți apeluri reale către API-ul Climatiq. Iată cum să-l structurați teste cu simulari robuste care simulează răspunsuri atât de succes, cât și de eșec.
# 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. Estimarea lotului și strategia de stocare în cache pentru Scala
Pentru aplicațiile care procesează mii de înregistrări (rapoarte lunare, analize istorice), combinația dintre apelurile API batch și memoria cache Redis devine esențială pentru a reduce atât timpii de procesare, cât și costurile 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. Alternative la Climatiq: comparație
Climatiq nu este singura opțiune. Iată o comparație a celor principale alternative pentru a vă ajuta să alegeți soluția potrivită:
| Soluţie | Forța principală | Limită | Caz de utilizare ideal | Plan gratuit |
|---|---|---|---|---|
| Climatiq | Peste 190.000 de factori, peste 40 de surse, API robust | 250 apeluri/luna in plan comunitar | Întreprindere, producție cu domenii multiple | 250 apeluri/luna |
| Interfață de carbon | API simplu, experiență excelentă pentru dezvoltatori | Mai puțini factori, concentrați-vă pe SUA | Startup-uri, comerț electronic, transport maritim | Da (limitat) |
| Factori de emisie deschisi | Gratuit, open source, întreținut de Climatiq | Numai seturi de date, fără API REST direct | Cercetare, acces la date brute | Da (set de date) |
| API-ul Watershed | Funcții de întreprindere pregătite pentru audit | Doar Enterprise, scump | Companii mari, raportare CSRD | No |
| Baza de date EPA FLIGHT | Liber, guvern american | Numai SUA, nu REST API | Raportare din SUA, cercetare | Da (set de date) |
| DB personalizat (DEFRA/IEA) | Control total, fără costuri API | Întreținere internă costisitoare | Organizații mari cu echipă dedicată | Da (date publice) |
Când să utilizați o bază de date personalizată
În unele cazuri, construirea unei baze de date interne a factorilor de emisie are sens. Iată o schemă PostgreSQL pentru a o gestiona:
-- 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;Când baza de date personalizată NU este alegerea corectă
Construirea și menținerea unei baze de date cu factori de emisie necesită: o persoană dedicată timp de 2-3 luni pentru construcția inițială, actualizări semestriale obligatorii (factorii se modifică în fiecare an), gestionarea versiunilor pentru pista de audit și acoperirea geografică care de multe ori lipsă pentru regiunile non-OCDE. Pentru majoritatea companiilor, Climatiq oferă un ROI superior chiar și luând în considerare costul API.
13. Studiu de caz: Platformă de comerț electronic cu Carbon Label
EcoShop Italia este o platformă de comerț electronic cu 50.000 de produse și 200.000 de comenzi/lună. Cerință: adăugați o etichetă de carbon la fiecare produs și afișați emisiile estimate la casă, cu opțiunea de compensare.
Arhitectură implementată
- Îmbogățirea catalogului de produse: Lucru de noapte care precalcula emisiile de producție pentru fiecare SKU folosind punctul final al lotului. Rezultate salvate în baza de date de produse cu un TTL de 30 de zile.
- Checkout în timp real: La casă, calculul dinamic al emisiilor livrare în funcție de greutatea coșului, regiunea de destinație și metoda aleasă. Latența țintă: sub 200 ms (cu cache Redis pentru expedieri standard).
- Vanzator Carbon Dashboard: Raport lunar pentru vânzători cu Domeniul de aplicare 3.4 (transport în amonte) și domeniul de aplicare estimat 3.11 (utilizarea produselor vândute).
# 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. Implementare și configurare sigură în producție
# 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_dbNu expuneți niciodată cheia API Climatiq
Cheia Climatiq API nu trebuie să fie niciodată în codul sursă, în jurnale, ca răspuns API-uri sau variabile JavaScript la nivelul clientului. Utilizați întotdeauna: variabile de mediu, manager de secrete (AWS Secrets Manager, HashiCorp Vault, Azure Key Vault) sau Kubernetes Secrets cu criptare în repaus. Rotiți cheia imediat dacă este expusă accidental.
15. Cele mai bune practici și anti-modele
Cele mai bune practici
-
Blocați data_version în producție: STATELE UNITE ALE AMERICII
"21"în loc de"^21"pentru a asigura reproductibilitatea calculelor. Actualizați în mod deliberat cu recalculare explicită. - Memorarea agresivă în cache: Factorii de emisie se modifică rar. Un cache de 24-168 de ore reduce dramatic apelurile API, mai ales pentru calcule repetitive (flotă lunară, catalog de produse).
- Întotdeauna pista de audit: Salvați fiecare calcul cu activity_id, factorul utilizat, versiunea datelor, marcajul de timp și sursa. Esențial pentru CSRD.
- Folosiți lotul atunci când este posibil: Un singur apel lot cu 100 de articole este mult mai eficient decât 100 de apeluri individuale. Respectă limitele ratei și reduce latența totală.
- Validați data_quality_flags: Climatiq raportează când a factorul are o calitate scăzută a datelor. Gestionați aceste cazuri cu factori alternativi sau note de incertitudine din raport.
- Raportare dublă pentru Scopul 2: Protocolul GHG cere atât metode bazate pe locație, cât și pe piață. Calculați și salvați ambele.
- Cache în memorie pentru TypeScript: Implementați o hartă cu TTL în clientul TypeScript pentru a evita apelurile redundante în aplicații Node.js cu viață lungă fără Redis.
Anti-modele de evitat
- Cheia API în front-end: Nu expuneți cheia Climatiq JavaScript pe partea clientului. Merge întotdeauna prin backend.
- Ignorați gazele constitutive: Pentru raportarea corectă a CSRD, CO₂, CH₄ și N₂O trebuie raportate separat în plus față de CO₂e total.
- Unități de măsură incorecte: Climatiq acceptă conversii între unități, dar trecerea de litri atunci când punctul final se așteaptă ca kWh produce rezultate plauzibil din punct de vedere numeric, dar greșit din punct de vedere științific. Mereu valabil.
- Lot > 100 articole: API-ul returnează eroarea 422. Implementați întotdeauna logica de fragmentare pentru seturi mari de date.
-
Ignorați potrivirea regiunii: Un factor pentru electricitate
fără a specifica utilizarea implicită globală a regiunii. Pentru Italia folosiți întotdeauna
"IT". -
Blocarea apelurilor sincrone: Nu utilizați niciodată biblioteci HTTP sincrone
în punctele finale asincrone. Utilizați întotdeauna
httpx.AsyncClientîn Python sauaxioscuasync/awaitîn TypeScript.
Concluzii și pașii următori
Climatiq rezolvă cea mai dificilă problemă a contabilizării automate a carbonului: cel baza de date a factorilor de emisie. Cu peste 190.000 de factori verificați, acoperire de 300 de regiuni și actualizări continue, vă permite să construiți sisteme de calcul GHG pregătite pentru producție în zile în loc de luni.
În acest articol am construit:
- Un Client Python robust cu reîncercare, cache Redis și gestionarea erorilor de tastare
- Un Client TypeScript/Node.js cu Axios și siguranță de tip complet pentru integrări frontend
- Calcule pentru Domeniul de aplicare 1 (flotă diesel/HVO), Domeniul de aplicare 2 (metoda duala) și principalele categorii Domeniul de aplicare 3
- Un Carbon Label API în timp real pentru comerțul electronic cu etichetă A-E și offset
- Testarea cu API simulat pentru medii CI/CD fără apeluri reale
- Strategii de procesare în loturi și stocare în cache pentru scara intreprinderii
Seria Green Software continuă
- Articolul precedent: CodeCarbon - Măsoară emisiile de cod rulează cu biblioteca open source Python.
- Articolul următor: Carbon Aware SDK - Cum să schimbați sarcinile de lucru în orele cu cea mai curată energie folosind prognoza intensității rețelei.
- Articol similar (Seria MLOps): Optimizați antrenarea modelelor ML pentru a reduce emisiile de carbon.
- Articol înrudit (Data & AI Business Series): Guvernarea datelor pentru Reliable AI - Cum să integrați valorile de sustenabilitate în catalogul de date.
Următorul pas practic este înscrierea în planul comunitar Climatiq (250 de apeluri gratuite/lună), explorați Explorator de date pentru a găsi factorii relevanți pentru industria dvs. și implementați calcularea Scopului 1 pentru cel mai simplu caz de utilizare ca prim reper.
Odată cu presiunile de reglementare ale CSRD (obligatoriu pentru marile companii din UE din 2025, extinsă la IMM-uri din 2026) și atenția din ce în ce mai mare a investitorilor pentru parametrii ESG, a avea infrastructură tehnică pentru contabilizarea automată a carbonului nu mai este un avantaj competitiv: este a necesitate operațională.