File, JSON, CSV ও HTTP
Data পড়া, লেখা আর অন্য API call করা
Backend এর কাজের বড় অংশ হলো data এক জায়গা থেকে আরেক জায়গায় নেওয়া: file থেকে পড়া, JSON বানানো, CSV export করা, অন্য কোনো API কে call করা। এই অধ্যায়ে এই চারটা কাজ শিখবো, আধুনিক আর নিরাপদ উপায়ে।
#Path: file এর ঠিকানা
পুরনো code এ os.path.join("data", "file.json") দেখবে। আধুনিক Python এ pathlib ব্যবহার করো:
from pathlib import Path
BASE_DIR = Path(__file__).parent # এই file যে folder এ আছে
DATA_DIR = BASE_DIR / "data" # / দিয়ে path জোড়া
DATA_DIR.mkdir(exist_ok=True) # folder না থাকলে বানাও
file = DATA_DIR / "expenses.json"
print(file.name) # expenses.json
print(file.suffix) # .json
print(file.exists())
for p in DATA_DIR.glob("*.json"): # সব json file
print(p)#Text file পড়া আর লেখা
from pathlib import Path
note = Path("note.txt")
# লেখা (পুরনোটা মুছে)
note.write_text("প্রথম লাইন\n", encoding="utf-8")
# শেষে যোগ করা
with note.open("a", encoding="utf-8") as f:
f.write("দ্বিতীয় লাইন\n")
# পুরোটা পড়া
print(note.read_text(encoding="utf-8"))
# লাইন ধরে পড়া (বড় file এর জন্য)
with note.open(encoding="utf-8") as f:
for line in f:
print(line.rstrip())| Mode | মানে |
|---|---|
"r" |
পড়া (default) |
"w" |
লেখা, আগের সব মুছে |
"a" |
শেষে যোগ করা |
"x" |
নতুন file, আগে থাকলে error |
"rb" / "wb" |
Binary (ছবি, PDF) |
#JSON
import json
from pathlib import Path
expenses = [
{"id": 1, "title": "দুপুরের খাবার", "amount": 18000, "category": "food"},
{"id": 2, "title": "রিকশা", "amount": 6000, "category": "transport"},
]
path = Path("expenses.json")
# Python → JSON file
path.write_text(json.dumps(expenses, ensure_ascii=False, indent=2), encoding="utf-8")
# JSON file → Python
loaded = json.loads(path.read_text(encoding="utf-8"))
print(loaded[0]["title"])ensure_ascii=False না দিলে বাংলা "দু..." হয়ে যাবে। পড়া যায়, কিন্তু মানুষ বুঝবে না।
#JSON আর Python type এর মিল
| JSON | Python |
|---|---|
object {} |
dict |
array [] |
list |
| string | str |
| number | int / float |
true / false |
True / False |
null |
None |
datetime, Decimal, নিজের class সরাসরি JSON এ যায় না। এর সহজ সমাধান Pydantic:
from datetime import datetime
from pydantic import BaseModel, TypeAdapter
class Expense(BaseModel):
id: int
title: str
amount: int
spent_at: datetime
items = [Expense(id=1, title="চা", amount=2000, spent_at=datetime.now())]
adapter = TypeAdapter(list[Expense])
raw = adapter.dump_json(items, indent=2) # bytes, datetime সহ
back = adapter.validate_json(raw) # আবার Expense list, যাচাইসহ#নিরাপদভাবে file save করা
Save করার মাঝখানে program crash করলে বা বিদ্যুৎ গেলে অর্ধেক লেখা file থেকে যায়, আর সব data নষ্ট হয়। সমাধান: আগে অস্থায়ী file এ লেখো, তারপর এক ধাপে নাম বদলাও:
import json
import os
from pathlib import Path
def save_json_atomic(path: Path, data) -> None:
tmp = path.with_suffix(".tmp")
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
os.replace(tmp, path) # এক ধাপে বদল: হয় পুরনো file, নয় পুরো নতুন fileDatabase এর transaction এর মতোই ধারণা: অর্ধেক অবস্থা কেউ দেখবে না।
#CSV: Excel এর বন্ধু
অফিসের মানুষ report চায় Excel এ। CSV বানালে Excel এ খোলা যায়:
import csv
from pathlib import Path
rows = [
{"date": "2026-10-01", "title": "বাজার", "amount": 1250.00},
{"date": "2026-10-02", "title": "বিদ্যুৎ বিল", "amount": 980.50},
]
path = Path("report.csv")
with path.open("w", newline="", encoding="utf-8-sig") as f: # -sig: Excel এ বাংলা ঠিক দেখাবে
writer = csv.DictWriter(f, fieldnames=["date", "title", "amount"])
writer.writeheader()
writer.writerows(rows)
with path.open(encoding="utf-8-sig") as f:
for row in csv.DictReader(f):
print(row["title"], float(row["amount"]))#HTTP: অন্য API কে call করা
httpx আধুনিক HTTP library (পুরনো requests এর মতো, কিন্তু async ও পারে):
import httpx
with httpx.Client(base_url="https://jsonplaceholder.typicode.com", timeout=10.0) as client:
# GET, query parameter সহ
r = client.get("/posts", params={"userId": 1})
r.raise_for_status() # 4xx/5xx হলে exception
posts = r.json()
print(len(posts), posts[0]["title"])
# POST, JSON body সহ
r = client.post("/posts", json={"title": "নতুন", "body": "লেখা", "userId": 1})
print(r.status_code, r.json())
# Header (যেমন token)
r = client.get("/users/1", headers={"Authorization": "Bearer abc123"})#Production মানের HTTP call এর নিয়ম
import httpx
import logging
logger = logging.getLogger(__name__)
class WeatherError(Exception):
pass
def get_temperature(client: httpx.Client, lat: float, lng: float) -> float:
try:
r = client.get(
"https://api.open-meteo.com/v1/forecast",
params={"latitude": lat, "longitude": lng, "current": "temperature_2m"},
)
r.raise_for_status()
except httpx.TimeoutException as e:
raise WeatherError("Weather service সময়মতো সাড়া দেয়নি") from e
except httpx.HTTPStatusError as e:
logger.warning("Weather API %s: %s", e.response.status_code, e.response.text[:200])
raise WeatherError("Weather service error") from e
data = r.json()
return data["current"]["temperature_2m"]- সবসময় timeout দাও। না দিলে অন্য server ঝুলে থাকলে তোমার server ও চিরকাল অপেক্ষা করবে, আর একটু পরে তোমার সব worker আটকে যাবে।
- একটা
Clientবারবার ব্যবহার করো। প্রতিবার নতুন client মানে নতুন TCP+TLS connection, যা ধীর। raise_for_status()ডাকো। না হলে 500 error এর HTML কে JSON ভেবে পড়তে গিয়ে অদ্ভুত error আসবে।- বাইরের error কে নিজের error এ বদলাও, যাতে উপরের code জানে কী ঘটেছে।
- ফেরত আসা data যাচাই করো (Pydantic দিয়ে), অন্যের API যেকোনো সময় বদলাতে পারে।
#তারিখ আর সময়
Backend এ সময় নিয়ে ভুল খুব সাধারণ। নিয়ম: database আর server এ সবসময় UTC রাখো, শুধু user কে দেখানোর সময় স্থানীয় সময়ে বদলাও।
from datetime import datetime, timedelta, timezone
from zoneinfo import ZoneInfo
now_utc = datetime.now(timezone.utc)
dhaka = now_utc.astimezone(ZoneInfo("Asia/Dhaka"))
print(dhaka.strftime("%d %B %Y, %I:%M %p")) # 29 October 2026, 10:15 AM
tomorrow = now_utc + timedelta(days=1)
iso = now_utc.isoformat() # API তে পাঠানোর জন্য: '2026-10-29T04:15:00+00:00'
parsed = datetime.fromisoformat(iso)#অনুশীলন
#নিজেকে যাচাই করো
`json.dumps` এ `ensure_ascii=False` কেন দিই?
না দিলে non-ASCII অক্ষর (বাংলা) দ এর মতো escape হয়ে যায়। JSON ঠিক থাকে, কিন্তু file খুলে মানুষ পড়তে পারে না।
HTTP call এ timeout না দিলে server এ কী হতে পারে?
অন্য server ঝুলে গেলে তোমার request চিরকাল অপেক্ষা করবে। প্রতিটা আটকে থাকা request একটা worker দখল করে রাখে, একসময় সব worker আটকে গিয়ে তোমার পুরো server সাড়া দেওয়া বন্ধ করে দেয়।
Atomic save কী, আর কেন দরকার?
আগে অস্থায়ী file এ পুরো data লিখে তারপর এক ধাপে (os.replace) আসল file এর জায়গায় বসানো। Save এর মাঝে crash হলেও আসল file হয় পুরনো অবস্থায় থাকে, নয় পুরো নতুন অবস্থায়। অর্ধেক লেখা নষ্ট file হয় না।
Database এ সময় কোন timezone এ রাখবে?
UTC তে। শুধু user কে দেখানোর সময় তার timezone (যেমন Asia/Dhaka) এ বদলাবে। এতে server কোন দেশে আছে বা user কোথায় আছে, তাতে হিসাব গুলিয়ে যায় না।
চিহ্ন দিলে sidebar আর রোডম্যাপে এই অধ্যায়ের station ভরে যাবে।