FastAPI बैकएंड डेवलपमेंट बूटकैंप · पाठ

स्कीमा रजिस्ट्री और Avro अनुबंध विकास

स्कीमा रजिस्ट्री से इवेंट अनुबंध लागू कीजिए और संगतता नियमों के अनुसार पेलोड विकसित कीजिए।

पाठ 2, कुल 4 में से13 चरण

स्कीमा रजिस्ट्री और Avro अनुबंध विकास, CoddyKit पर FastAPI बैकएंड डेवलपमेंट बूटकैंप का एक निःशुल्क पाठ है। यह 4 में से 2वाँ पाठ है। इस अध्ययन पथ के 3 तक कोई भी पाठ पूरा पढ़ना निःशुल्क है — इसके बाद CoddyKit PRO हर पाठ अनलॉक करता है, साथ ही अंतर्निर्मित कोड संपादक और चौबीसों घंटे एआई शिक्षक के साथ व्यावहारिक अभ्यास भी उपलब्ध कराता है। यह FastAPI बैकएंड डेवलपमेंट बूटकैंप सीखने के मार्ग का हिस्सा है और आपकी प्रगति वेब तथा CoddyKit ऐप पर सिंक होती रहती है। FastAPI बैकएंड डेवलपमेंट बूटकैंप पाठ्यक्रम में कुल 4 पाठ शामिल हैं।

इवेंट कॉन्ट्रैक्ट के लिए रजिस्ट्री क्यों आवश्यक है

इवेंट-आधारित FastAPI बैकएंड में, आपकी सेवा Kafka या Pulsar पर इवेंट प्रकाशित करती है और कई स्वतंत्र उपभोक्ता उन्हें पढ़ते हैं। इवेंट पेलोड एक कॉन्ट्रैक्ट होता है: उत्पादकों और उपभोक्ताओं को फ़ील्ड नामों, प्रकारों और संरचना पर सहमत होना चाहिए।

यदि उत्पादक user_id का नाम बदलकर userId कर दे, तो हर उपभोक्ता चुपचाप विफल हो जाता है।
साधारण JSON में कोई अनिवार्य आकार नहीं होता, इसलिए टाइपो सीधे प्रोडक्शन तक पहुँच जाता है।

स्कीमा रजिस्ट्री संस्करणयुक्त स्कीमा को केंद्रीय रूप से संग्रहीत करती है और सहमत कॉन्ट्रैक्ट का उल्लंघन करने वाले संदेशों को अस्वीकार करती है। इस तरह डेटा सुरक्षित रखते हुए टीमों को अलग-अलग काम करने की सुविधा मिलती है।

Avro: एक संक्षिप्त, स्कीमा-प्रथम प्रारूप

Apache Avro स्कीमा रजिस्ट्री के साथ उपयोग किया जाने वाला सबसे सामान्य सीरियलाइज़ेशन प्रारूप है। प्रत्येक रिकॉर्ड का वर्णन JSON स्कीमा द्वारा किया जाता है, और बाइनरी पेलोड में स्वयं फ़ील्ड नाम नहीं, केवल मान होते हैं, जिससे यह संक्षिप्त रहता है।

स्कीमा name, type, fields और वैकल्पिक default मान निर्धारित करता है।
बाइट्स को डिकोड करने के लिए पाठकों को स्कीमा की आवश्यकता होती है; रजिस्ट्री का अस्तित्व ठीक इसी कारण है।

नीचे OrderCreated इवेंट के लिए एक न्यूनतम Avro स्कीमा दिया गया है।

order_created_schema = {
    "type": "record",
    "name": "OrderCreated",
    "namespace": "com.shop.events",
    "fields": [
        {"name": "order_id", "type": "string"},
        {"name": "user_id", "type": "string"},
        {"name": "amount_cents", "type": "long"},
        {"name": "currency", "type": "string"},
    ],
}

print(order_created_schema["name"], "has", len(order_created_schema["fields"]), "fields")

fastavro से रिकॉर्ड को सीरियलाइज़ करना

शुद्ध-Python fastavro लाइब्रेरी आपको किसी ब्रोकर के बिना Avro रिकॉर्ड को एन्कोड और डिकोड करने देती है। यही बाइट लेआउट आपका उत्पादक Kafka पर भेजेगा।

parse_schema स्कीमा को एक बार मान्य करता है।
schemaless_writer बाइनरी बॉडी लिखता है; schemaless_reader उसे डिकोड करता है।

ध्यान दें कि एन्कोड किए गए बाइट्स में केवल मान होते हैं, फ़ील्ड नाम नहीं।

import io
from fastavro import parse_schema, schemaless_writer, schemaless_reader

schema = parse_schema({
    "type": "record",
    "name": "OrderCreated",
    "fields": [
        {"name": "order_id", "type": "string"},
        {"name": "amount_cents", "type": "long"},
    ],
})

record = {"order_id": "o-123", "amount_cents": 4999}

buf = io.BytesIO()
schemaless_writer(buf, schema, record)
encoded = buf.getvalue()
print("encoded bytes:", encoded)

buf.seek(0)
decoded = schemaless_reader(buf, schema)
print("decoded:", decoded)

Confluent वायर प्रारूप

जब आप रजिस्ट्री के माध्यम से प्रकाशित करते हैं, तो मान केवल Avro नहीं होता। Confluent का सीरियलाइज़र 5-बाइट हेडर जोड़ता है, ताकि उपभोक्ताओं को पता हो कि कौन-सा स्कीमा प्राप्त करना है।

बाइट 0: एक मैजिक बाइट, हमेशा 0x00।
बाइट 1-4: big-endian 4-बाइट स्कीमा ID।
शेष बाइट्स: बिना स्कीमा वाला Avro बॉडी।

उपभोक्ता ID पढ़ता है, रजिस्ट्री से उसी स्कीमा संस्करण को डाउनलोड करता है और बॉडी को डिकोड करता है। इसी तरह पुराने और नए पेलोड एक ही टॉपिक पर साथ रह सकते हैं।

import struct

MAGIC = 0
schema_id = 42
avro_body = b"\x0co-1234\x9eL"  # pretend Avro bytes

frame = struct.pack(">bI", MAGIC, schema_id) + avro_body
print("wire bytes:", frame)

magic, sid = struct.unpack(">bI", frame[:5])
print("magic:", magic, "schema_id:", sid)
print("body:", frame[5:])

FastAPI स्टार्टअप से स्कीमा रजिस्टर करना

एक स्वच्छ तरीका यह है कि एप्लिकेशन स्टार्टअप पर रजिस्ट्री के REST API का उपयोग करके अपने उत्पादक का स्कीमा एक बार रजिस्टर करें। रजिस्ट्री एक स्थिर स्कीमा ID लौटाती है, जिसे आप हर संदेश के लिए पुनः उपयोग करते हैं।

डिफ़ॉल्ट रूप से विषय <topic>-value परंपरा का पालन करते हैं।
एक समान स्कीमा को रजिस्टर करना इडेम्पोटेंट है: आपको वही ID वापस मिलती है।

यह स्निपेट Confluent-संगत रजिस्ट्री पर Avro स्कीमा पोस्ट करता है।

import json
import httpx

REGISTRY_URL = "http://schema-registry:8081"

async def register_schema(subject: str, avro_schema: dict) -> int:
    payload = {"schema": json.dumps(avro_schema)}
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            f"{REGISTRY_URL}/subjects/{subject}/versions",
            json=payload,
            headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
        )
        resp.raise_for_status()
        return resp.json()["id"]

# Called inside FastAPI's lifespan startup:
# schema_id = await register_schema("orders-value", order_created_schema)

संगतता मोड: मुख्य निर्णय

रजिस्ट्री प्रत्येक विषय के लिए विकास नीति लागू करती है। आपके द्वारा चुना गया मोड तय करता है कि कौन-से स्कीमा परिवर्तन अनुमत हैं और आपका अपग्रेड क्रम निर्धारित करता है।

BACKWARD (डिफ़ॉल्ट): नया स्कीमा पिछले स्कीमा द्वारा लिखे गए डेटा को पढ़ सकता है। पहले उपभोक्ताओं को अपग्रेड करें।
FORWARD: पिछला स्कीमा नए स्कीमा द्वारा लिखे गए डेटा को पढ़ सकता है। पहले उत्पादकों को अपग्रेड करें।
FULL: दोनों दिशाएँ लागू होती हैं। क्रम महत्वपूर्ण नहीं है।
*_TRANSITIVE: जाँच केवल नवीनतम नहीं, बल्कि सभी पिछले संस्करणों के विरुद्ध चलती है।

अधिकांश टीमें डिफ़ॉल्ट रूप से BACKWARD चुनती हैं, क्योंकि उपभोक्ता आमतौर पर उत्पादकों से पीछे रहते हैं।

पिछली संगतता वाला परिवर्तन: डिफ़ॉल्ट मान के साथ फ़ील्ड जोड़ना

BACKWARD संगतता के अंतर्गत आप केवल तभी फ़ील्ड जोड़ सकते हैं, जब उसका डिफ़ॉल्ट मान हो। पुराने संदेश को पढ़ने वाला नया स्कीमा उस डिफ़ॉल्ट मान को स्वतः भर देता है; हटाए गए फ़ील्ड के लिए भी आवश्यक है कि पुराने फ़ील्ड का कोई डिफ़ॉल्ट मान रहा हो।

discount_cents को default: 0 के साथ जोड़ना सुरक्षित है।
इसे डिफ़ॉल्ट मान के बिना जोड़ना अस्वीकार कर दिया जाता है, क्योंकि पुराने रिकॉर्ड में देने के लिए कोई मान नहीं होता।

नीचे दिया गया नया संस्करण ऑर्डर इवेंट को सुरक्षित रूप से विकसित करता है।

order_v2 = {
    "type": "record",
    "name": "OrderCreated",
    "namespace": "com.shop.events",
    "fields": [
        {"name": "order_id", "type": "string"},
        {"name": "user_id", "type": "string"},
        {"name": "amount_cents", "type": "long"},
        {"name": "currency", "type": "string"},
        # NEW field is backward compatible ONLY because of the default
        {"name": "discount_cents", "type": "long", "default": 0},
    ],
}

print("fields in v2:", [f["name"] for f in order_v2["fields"]])

नए स्कीमा से पुराने बाइट्स पढ़ना

Avro लेखक स्कीमा (एन्कोड करने के लिए प्रयुक्त) और पाठक स्कीमा (डिकोड करने के लिए प्रयुक्त) के बीच के अंतर को सुलझाता है। जब पाठक के पास डिफ़ॉल्ट मान वाला नया फ़ील्ड होता है, तो पुराने बाइट्स को डिकोड करते समय वह डिफ़ॉल्ट मान अपने-आप जोड़ दिया जाता है।

schemaless_reader को दोनों स्कीमा पाठक और लेखक के रूप में दें।
अनुपस्थित discount_cents अपने डिफ़ॉल्ट 0 के रूप में दिखाई देता है।

यही वह कारण है जिससे BACKWARD विकास प्रोडक्शन में बिना रुकावट के होता है।

import io
from fastavro import parse_schema, schemaless_writer, schemaless_reader

writer = parse_schema({
    "type": "record", "name": "OrderCreated",
    "fields": [{"name": "order_id", "type": "string"},
               {"name": "amount_cents", "type": "long"}],
})
reader = parse_schema({
    "type": "record", "name": "OrderCreated",
    "fields": [{"name": "order_id", "type": "string"},
               {"name": "amount_cents", "type": "long"},
               {"name": "discount_cents", "type": "long", "default": 0}],
})

buf = io.BytesIO()
schemaless_writer(buf, writer, {"order_id": "o-9", "amount_cents": 1500})
buf.seek(0)
out = schemaless_reader(buf, writer, reader)
print(out)  # discount_cents filled from default

रजिस्ट्री द्वारा अस्वीकार किए जाने वाले विघटनकारी परिवर्तन

कुछ संपादन कभी संगत नहीं हो सकते और रजिस्ट्री की संगतता जाँच (एक पूर्व-जाँच POST .../compatibility/subjects/<s>/versions/latest) is_compatible: false लौटाएगी।

किसी फ़ील्ड का नाम बदलना (यह उपनामों के बिना जोड़ने और हटाने में बदल जाता है)।
किसी प्रकार को असंगत रूप से बदलना, जैसे string से long।
BACKWARD के अंतर्गत बिना डिफ़ॉल्ट मान वाला आवश्यक फ़ील्ड जोड़ना।

सुरक्षित रूप से नाम बदलने के लिए Avro के उपनामों का उपयोग करें, ताकि पाठक पुराने नाम को नए नाम से मैप कर सके।

# Safe rename using aliases: old name "user_id" -> new "customer_id"
renamed = {
    "type": "record",
    "name": "OrderCreated",
    "fields": [
        {"name": "order_id", "type": "string"},
        {
            "name": "customer_id",
            "type": "string",
            "aliases": ["user_id"],
        },
        {"name": "amount_cents", "type": "long"},
    ],
}

for f in renamed["fields"]:
    print(f["name"], f.get("aliases", []))

डिप्लॉय करने से पहले CI में संगतता जाँचना

विघटनकारी परिवर्तनों को ब्रोकर तक पहुँचने से पहले पकड़ें: अपनी CI पाइपलाइन से रजिस्ट्री के संगतता एंडपॉइंट को कॉल करें। यदि प्रस्तावित स्कीमा असंगत है, तो बिल्ड विफल कर दें।

यह Kafka के माध्यम से एक भी संदेश चलाए बिना हर उपभोक्ता की सुरक्षा करता है।
इसे उसी जॉब में एक चरण के रूप में चलाएँ जो आपकी FastAPI इमेज बनाती है।

हेल्पर केवल तभी True लौटाता है, जब रजिस्ट्री नए संस्करण को मंज़ूरी देती है।

import json
import httpx

REGISTRY_URL = "http://schema-registry:8081"

async def is_compatible(subject: str, new_schema: dict) -> bool:
    url = f"{REGISTRY_URL}/compatibility/subjects/{subject}/versions/latest"
    async with httpx.AsyncClient() as client:
        resp = await client.post(
            url,
            json={"schema": json.dumps(new_schema)},
            headers={"Content-Type": "application/vnd.schemaregistry.v1+json"},
        )
        resp.raise_for_status()
        return resp.json()["is_compatible"]

# In CI:
#   ok = await is_compatible("orders-value", order_v2)
#   if not ok: raise SystemExit("Schema change is incompatible")

confluent-kafka के साथ उत्पादन और उपभोग

प्रोडक्शन में आप वायर प्रारूप और रजिस्ट्री लुकअप को सीरियलाइज़र के भरोसे छोड़ते हैं। confluent-kafka का AvroSerializer स्कीमा रजिस्टर करता है, ID जोड़ता है और बॉडी को एन्कोड करता है; AvroDeserializer इसका उल्टा करता है।

सीरियलाइज़र स्कीमा ID को कैश करता है, इसलिए रजिस्ट्री तक बहुत कम बार पहुँचा जाता है।
उपभोक्ता पारदर्शी रूप से उस लेखक स्कीमा को प्राप्त कर लेते हैं, जिसके साथ प्रत्येक संदेश एन्कोड किया गया था।

यही आपके FastAPI इवेंट प्रकाशक और डाउनस्ट्रीम सेवाओं के बीच का जोड़ है।

from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka import Producer

sr = SchemaRegistryClient({"url": "http://schema-registry:8081"})

schema_str = '''
{"type":"record","name":"OrderCreated",
 "fields":[{"name":"order_id","type":"string"},
          {"name":"amount_cents","type":"long"}]}
'''

serializer = AvroSerializer(sr, schema_str)
producer = Producer({"bootstrap.servers": "kafka:9092"})

# producer.produce(topic="orders",
#   value=serializer({"order_id": "o-1", "amount_cents": 999}, ctx))

त्वरित जाँच: संगतता मोड चुनना

आपकी टीम को Kafka इवेंट में एक नया वैकल्पिक फ़ील्ड जोड़ना है, और आप उत्पादक के साथ ठीक उसी समय हर उपभोक्ता को फिर से डिप्लॉय नहीं कर सकते। डिप्लॉयमेंट में उपभोक्ता आमतौर पर उत्पादकों से पीछे रहते हैं।

इंटरैक्टिव पाठ में इसका उत्तर दें

पुनरावलोकन: सुरक्षित रूप से विकसित होने वाले कॉन्ट्रैक्ट

अब आप जानते हैं कि इवेंट-आधारित FastAPI बैकएंड में इवेंट कॉन्ट्रैक्ट को कैसे लागू और विकसित करना है।

स्कीमा रजिस्ट्री संस्करणयुक्त स्कीमा संग्रहीत करती है और Confluent वायर प्रारूप (मैजिक बाइट + 4-बाइट ID + Avro बॉडी) में अंतर्निहित स्कीमा ID उपलब्ध कराती है।
Avro लेखक और पाठक स्कीमा को अलग रखता है तथा डिफ़ॉल्ट मानों और उपनामों के माध्यम से अंतर सुलझाता है।
BACKWARD (सामान्य डिफ़ॉल्ट) का अर्थ है कि नए स्कीमा पुराने डेटा को पढ़ते हैं: फ़ील्ड केवल डिफ़ॉल्ट मानों के साथ जोड़ें और पहले उपभोक्ताओं को अपग्रेड करें।
FORWARD में पहले उत्पादकों को अपग्रेड किया जाता है; FULL किसी भी क्रम की अनुमति देता है; TRANSITIVE रूपांतर सभी पिछले संस्करणों की जाँच करते हैं।
विघटनकारी परिवर्तनों को Kafka या Pulsar तक पहुँचने से रोकने के लिए रजिस्ट्री की संगतता जाँच CI में चलाएँ।

अपने स्कीमा को कोड की तरह मानें: उनके संस्करण बनाएँ, उनकी समीक्षा करें और कॉन्ट्रैक्ट की सुरक्षा रजिस्ट्री को करने दें।

शुरुआत निःशुल्क

एआई शिक्षक के साथ FastAPI बैकएंड डेवलपमेंट बूटकैंप सीखें — निःशुल्क

अपने ब्राउज़र में वास्तविक कोड लिखें और चलाएँ, चौबीसों घंटे एआई शिक्षक से तुरंत सहायता पाएँ, और वेब या ऐप पर वहीं से शुरू करें जहाँ आपने छोड़ा था।

पाठ्यक्रम: 21
पाठ: 84

निःशुल्क सीखना शुरू करें पूरा पाठ्यक्रम देखें

अक्सर पूछे जाने वाले प्रश्न

क्या “स्कीमा रजिस्ट्री और Avro अनुबंध विकास” पाठ निःशुल्क है?

हाँ — FastAPI बैकएंड डेवलपमेंट बूटकैंप अध्ययन पथ के 3 तक कोई भी पाठ, जिसमें “स्कीमा रजिस्ट्री और Avro अनुबंध विकास” भी शामिल है, यहाँ वेब पर पूरा पढ़ना निःशुल्क है। इसके बाद CoddyKit PRO हर पाठ अनलॉक करता है, साथ ही अंतर्निर्मित कोड संपादक और चौबीसों घंटे एआई शिक्षक के साथ इंटरैक्टिव अभ्यास भी उपलब्ध कराता है। FastAPI बैकएंड डेवलपमेंट बूटकैंप पाठ्यक्रम में कुल 4 पाठ शामिल हैं।

“स्कीमा रजिस्ट्री और Avro अनुबंध विकास” में मैं क्या सीखूँगा?

स्कीमा रजिस्ट्री से इवेंट अनुबंध लागू कीजिए और संगतता नियमों के अनुसार पेलोड विकसित कीजिए। आप ब्राउज़र में सीधे चलाए जाने वाले व्यावहारिक कोड के साथ FastAPI बैकएंड डेवलपमेंट बूटकैंप का अभ्यास करते हैं, और पाठ पूरा करते समय 24/7 एआई ट्यूटर आपके प्रश्नों के उत्तर देता है।

क्या FastAPI बैकएंड डेवलपमेंट बूटकैंप शुरू करने के लिए मुझे किसी अनुभव की आवश्यकता है?

पहले के अनुभव की आवश्यकता नहीं है। CoddyKit पर FastAPI बैकएंड डेवलपमेंट बूटकैंप शुरुआती से लेकर उन्नत शिक्षार्थियों तक सभी के लिए व्यवस्थित किया गया है, इसलिए आप यहीं से या शुरुआत से सीखना शुरू कर सकते हैं और अपनी गति से आगे बढ़ सकते हैं। यह 4 में से 2वाँ पाठ है।

“स्कीमा रजिस्ट्री और Avro अनुबंध विकास” पाठ पूरा करने में कितना समय लगता है?

CoddyKit का अधिकांश पाठ लगभग 5–10 मिनट में पूरा हो जाता है। हर पाठ छोटा और संवादात्मक है, इसलिए आप लगातार प्रगति करते हैं और वेब या ऐप पर वहीं से सीखना जारी रख सकते हैं जहाँ आपने छोड़ा था।

क्या मैं इस FastAPI बैकएंड डेवलपमेंट बूटकैंप पाठ में कोड लिख और चला सकता हूँ?

हाँ। हर FastAPI बैकएंड डेवलपमेंट बूटकैंप पाठ में एक अंतर्निर्मित कोड संपादक शामिल है, जिससे आप सीधे अपने ब्राउज़र में वास्तविक कोड लिख और चला सकते हैं और तुरंत एआई प्रतिक्रिया पा सकते हैं—स्थानीय सेटअप की आवश्यकता नहीं है।

इस पाठ्यक्रम के सभी पाठ

Kafka इवेंट का अतुल्यकालिक उत्पादन और उपभोग
स्कीमा रजिस्ट्री और Avro अनुबंध विकास
लेन-देनात्मक आउटबॉक्स पैटर्न
इडेम्पोटेंट उपभोक्ता और ठीक-एक-बार अर्थविज्ञान

← FastAPI बैकएंड डेवलपमेंट बूटकैंप पर वापस जाएँ