MongoDB의 변경 스트림을 사용하면 애플리케이션이 실시간 데이터 변경에 즉시 반응할 수 있습니다. 이 블로그 게시물에서는 이론에 너무 깊이 들어가지 않고도 Python을 사용하여 변경 스트림을 설정하고 사용하는 방법을 보여 드리겠습니다. 먼저 삽입에 초점을 맞춘 다음 이를 다른 이벤트 유형으로 확장하여 데이터베이스 이벤트를 수신하는 간단한 프로그램을 만들어 보겠습니다.
변경 스트림을 사용하면 앱이 삽입이나 업데이트와 같은 특정 데이터베이스 이벤트를 수신하고 즉시 응답할 수 있습니다. 사용자가 자신의 프로필을 업데이트하는 시나리오를 상상해 보세요. 변경 스트림을 사용하면 사용자가 페이지를 새로 고칠 필요 없이 앱 전체에 이 변경 사항을 즉시 반영할 수 있습니다. 이 기능 이전에는 지속적으로 데이터베이스를 폴링하거나 MongoDB Oplog 테일링과 같은 복잡한 방법을 사용해야 했습니다. 변경 스트림은 보다 사용자 친화적인 API를 제공하여 이를 단순화합니다.
인보이스를 업로드하는 API가 있다고 가정해 보겠습니다. 흐름은 고객이 MongoDB에 송장 이미지를 업로드한 다음 AI로 정보를 추출하고 송장을 업데이트하는 것입니다. 송장 업로드 코드의 예는 다음과 같습니다.
from pymongo import MongoClient class MongoDatabase: def __init__(self, config_path: str): # Load the YAML configuration file using the provided utility function self.config_path = config_path self.config = read_config(path=self.config_path) # Initialize MongoDB connection self.client = MongoClient(self.config['mongodb']['uri']) self.db = self.client[self.config['mongodb']['database']] self.collection = self.db[self.config['mongodb']['collection']] def create_document(self, data: Dict[str, Any]) -> str: # Insert a new document and return the automatically generated document ID result = self.collection.insert_one(data) return str(result.inserted_id) def update_document_by_id(self, document_id: str, data: Dict[str, Any]): try: self.collection.update_one({"_id": document_id}, {"$set": data}) except PyMongoError as e: print(f"Error updating document: {e}")
먼저 만일을 대비해 pymongo를 클래스 안에 포장하겠습니다 :))
@app.post("/api/v1/invoices/upload") async def upload_invoice(request: Request): try: # Parse JSON body body = await request.json() img = body.get("img") user_uuid = body.get("user_uuid") if not img or not is_base64(img): return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"status": "error", "message": "Base64 image is required"}, ) # Generate invoice UUID current_time = datetime.now(timezone.utc) img = valid_base64_image(img) invoice_document = { "invoice_type": None, "created_at": current_time, "created_by": user_uuid, "last_modified_at": None, "last_modified_by": None, "status": "not extracted", "invoice_image_base64": img, "invoice_info": {} } invoice_uuid = mongo_db.create_document(invoice_document) print('Result saved to MongoDB:', invoice_uuid) mongo_db.update_document_by_id(invoice_uuid, {"invoice_uuid": invoice_uuid}) return JSONResponse( status_code=status.HTTP_201_CREATED, content={"invoice_uuid": invoice_uuid, "message": "Upload successful"} ) except Exception as e: # Handle errors return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"status": "error", "message": str(e)} )
합리적인 질문은 다음과 같습니다. 업데이트하기 전에 AI 모델이 이미지를 처리할 때까지 기다리지 않으시겠습니까? 문제는 처리하는 데 약 4~5분 정도 소요된다는 점이며, 사용자 경험에 영향을 미치고 싶지 않습니다.
또 다른 옵션은 Kafka를 사용하는 것입니다. 이미지를 Kafka 주제에 게시하면 다른 서비스가 데이터를 처리할 수 있습니다.
장점:
단점:
다음은 Kafka를 사용하여 송장 업로드 프로세스를 처리하는 방법을 보여주는 기본 구현입니다.
사용자가 API 엔드포인트를 통해 송장을 업로드합니다. 송장 이미지는 MongoDB에 저장되며 추가 처리를 위해 메시지가 Kafka 주제로 전송됩니다.
from kafka import KafkaProducer import json from fastapi import FastAPI, Request, status from fastapi.responses import JSONResponse from datetime import datetime, timezone app = FastAPI() producer = KafkaProducer( bootstrap_servers=['localhost:9092'], value_serializer=lambda v: json.dumps(v).encode('utf-8') ) @app.post("/api/v1/invoices/upload") async def upload_invoice(request: Request): try: body = await request.json() img = body.get("img") user_uuid = body.get("user_uuid") if not img or not is_base64(img): return JSONResponse( status_code=status.HTTP_400_BAD_REQUEST, content={"status": "error", "message": "Base64 image is required"}, ) current_time = datetime.now(timezone.utc) img = valid_base64_image(img) invoice_document = { "invoice_type": None, "created_at": current_time, "created_by": user_uuid, "status": "not extracted", "invoice_image_base64": img, } # Save the document to MongoDB invoice_uuid = mongo_db.create_document(invoice_document) mongo_db.update_document_by_id(invoice_uuid, {"invoice_uuid": invoice_uuid}) # Send a message to Kafka topic producer.send('invoice_topic', invoice_document) producer.flush() return JSONResponse( status_code=status.HTTP_201_CREATED, content={"message": "Invoice upload received and will be processed"} ) except Exception as e: return JSONResponse( status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, content={"status": "error", "message": str(e)} )
Kafka 소비자는voice_topic을 듣습니다. 메시지를 받으면 송장을 처리하고(예: 이미지에서 정보 추출) MongoDB에서 해당 문서를 업데이트합니다.
from kafka import KafkaConsumer import json consumer = KafkaConsumer( 'invoice_topic', bootstrap_servers=['localhost:9092'], value_deserializer=lambda m: json.loads(m.decode('utf-8')) ) for message in consumer: invoice_document = message.value # Process the invoice, extract information, and update the document in MongoDB invoice_uuid = invoice_document["_id"] extracted_data = extract_invoice_data(invoice_document["invoice_image_base64"]) mongo_db.update_document_by_id(invoice_uuid, { "invoice_info": extracted_data, "status": "extracted" }) print(f"Processed and updated invoice: {invoice_uuid}")
흐름 요약:
와, 제가 이걸 직접 썼다니 믿을 수가 없네요! 실제로 관련된 노력을 강조합니다. MongoDB, Kafka, Invoice 서비스라는 세 가지 서비스를 관리하고 구성하는 복잡성도 고려하지 않은 것입니다.
다음은 변경 스트림에 의해 트리거되는 송장 처리를 처리하는 추가 메서드와 기능을 포함하여 MongoDB 변경 스트림을 보여주기 위해 Markdown으로 다시 작성된 전체 코드입니다.
문서 생성 및 변경 스트림 수신과 같은 데이터베이스 작업을 처리하는 MongoDB 래퍼 클래스를 만드는 것부터 시작하겠습니다.
from pymongo import MongoClient from pymongo.errors import PyMongoError from typing import Dict, Any import threading import yaml class MongoDatabase: # Same code as before # def process_invoice(self, invoice_document: Dict[str, Any]): """Process the invoice by extracting data and updating the document in MongoDB.""" try: # Simulate extracting information from the invoice image extracted_data = extract_invoice_data(invoice_document["invoice_image_base64"]) invoice_uuid = invoice_document["_id"] # Update the invoice document with the extracted data self.update_document_by_id(invoice_uuid, {"invoice_info": extracted_data, "status": "extracted"}) print(f"Processed and updated invoice: {invoice_uuid}") except Exception as e: print(f"Error processing invoice: {str(e)}") def start_change_stream_listener(self): """Start listening to the change stream for the collection.""" def listen(): try: with self.collection.watch() as stream: for change in stream: if change['operationType'] == 'insert': invoice_document = change['fullDocument'] print(f"New invoice detected: {invoice_document['_id']}") self.process_invoice(invoice_document) except PyMongoError as e: print(f"Change stream error: {str(e)}") # Start the change stream listener in a separate thread listener_thread = threading.Thread(target=listen, daemon=True) listener_thread.start()
쉽게 하기 위해 MongoDatabase 클래스 내에 process_invoice를 추가합니다. 하지만 다른 곳에 맡겨야 합니다
업로드 API는 원본과 같아야 합니다.
mongo_db = MongoDatabase(config_path='path_to_your_config.yaml') mongo_db.start_change_stream_listener() @app.post("/api/v1/invoices/upload") async def upload_invoice(request: Request): try: # Parse JSON body body = await request.json() # same code as before
흐름 요약:
MongoDB 변경 스트림을 사용하면 데이터베이스의 실시간 변경 사항을 효율적으로 처리할 수 있습니다. 이 예를 확장하면 업데이트 및 삭제와 같은 다양한 이벤트를 처리하여 애플리케이션의 반응성과 반응성을 높일 수 있습니다.
위 내용은 MongoDB 변경 스트림 및 Python을 사용한 실시간 데이터 처리의 상세 내용입니다. 자세한 내용은 PHP 중국어 웹사이트의 기타 관련 기사를 참조하세요!