python event bus

f1_event_bus.py

from collections import defaultdict
from threading import Lock

class EventBus:
    def __init__(self):
        self.events = defaultdict(list)
        self.lock = Lock()

    def subscribe(self, event, callback):
        with self.lock:
            self.events[event].append(callback)

    def unsubscribe(self, event, callback):
        with self.lock:
            self.events[event].remove(callback)

    def emit(self, event, *args, **kwargs):
        with self.lock:
            for callback in self.events[event]:
                callback(*args, **kwargs)

f2_generator.py

from event_bus import EventBus

event_bus = EventBus()
event_bus.emit('myEvent', {'data': 'hello'})

f3_listener.py

from event_bus import EventBus

event_bus = EventBus()

def handle_event(event_data):
    print(event_data)

event_bus.subscribe('myEvent', handle_event)

f3a_listener_async.py

import asyncio
#if async
async def handle_event(event_data):
    # Some async operation
    result = await some_async_operation()
    print(result)

event_bus.subscribe('myEvent', handle_event)

f4_event_bus_chunk_file_and_send_event.py

import json
from collections import defaultdict
from threading import Lock

class EventBus:
    def __init__(self):
        self.events = defaultdict(list)
        self.lock = Lock()

    def subscribe(self, event, callback):
        with self.lock:
            self.events[event].append(callback)

    def unsubscribe(self, event, callback):
        with self.lock:
            self.events[event].remove(callback)

    def emit(self, event, *args, **kwargs):
        with self.lock:
            for callback in self.events[event]:
                callback(*args, **kwargs)

event_bus = EventBus()

# Read the file in chunks
with open('large_file.json', 'r') as f:
    while True:
        chunk = f.read(1024 * 1024) # read 1MB at a time
        if not chunk:
            break
        # Send the chunk to the event bus
        event_bus.emit('dataChunk', json.loads(chunk))

f4s_event_bus_chunk_file_and_send_SQS.py

import json
import boto3 #pip install boto3

sqs = boto3.client('sqs')
queue_url = 'YOUR_QUEUE_URL'

# Read the file in chunks
with open('large_file.json', 'r') as f:
    while True:
        chunk = f.read(1024 * 1024) # read 1MB at a time
        if not chunk:
            break
        # Send the chunk to SQS
        sqs.send_message(QueueUrl=queue_url, MessageBody=json.dumps(chunk))

f5_mongodb_chunk.py

from pymongo import MongoClient

# Connect to MongoDB
client = MongoClient("mongodb://localhost:27017/")

# Select the database
db = client["mydb"]

# Enable sharding on the collection
db.admin.command({"enableSharding": "mydb"})

# Choose the shard key
shard_key = {"user_id": "hashed"}

# Shard the collection
db.admin.command({"shardCollection": "mydb.bigCollection", "key": shard_key})
db.admin.command({"shardCollection": "mydb.bigCollection", "key": shard_key, "numInitialChunks":10})