neonize

repository·master·Indexed 19 days ago

https://github.com/krypton-byte/neonize

A high-performance Python library for WhatsApp automation built on the Go-based Whatsmeow library. It provides an event-driven API for sending and receiving messages, managing groups, handling media in real-time, and creating interactive polls. Neonize supports both synchronous (NewClient) and asynchronous (NewAClient) workflows, and includes a ClientFactory for managing multiple WhatsApp sessions. Supported database backends for session storage include SQLite, PostgreSQL, and in-memory options.

Tokens
51K
Snippets
186
Records
231
Agent score
64%

What's inside neonize

  1. Overview of Neonize API Modules

    master

    The Neonize API is organized into several key modules that handle different aspects of WhatsApp automation:

    • Core Client (neonize.client): Provides the synchronous client interface for managing connections and messaging.
    • Async Client (neonize.aioze.client): Provides an asynchronous interface for non-blocking automation tasks.
    • Events (neonize.events): Contains event types and handlers for reacting to WhatsApp lifecycle events (e.g., receiving messages).
    • Types (neonize.types): Defines the data models and types used throughout the library.
    • Utilities (neonize.utils): Provides helper functions for common tasks.
    • Exceptions (neonize.exc): Contains the hierarchy of exception classes for error handling.
  2. Overview of Neonize features

    master

    Neonize is a Python library for WhatsApp automation built on the Go-based whatsmeow library. It provides an event-driven architecture with support for both synchronous and asynchronous APIs.

    Core Capabilities

    • Messaging: Send/receive text, handle media (images, video, docs, audio), and manage group operations.
    • Real-time Events: Handle message receipts, status tracking, presence, and typing indicators.
    • Advanced Features: End-to-end encryption, contact retrieval, call event handling, polls, interactive messages, and newsletter/channel support.
    • Infrastructure: Built-in logging and support for SQLite and PostgreSQL databases.
  3. How the Async Event Loop architecture works

    master

    Neonize handles events from Go callbacks running on separate OS threads. To bridge these to Python, it uses asyncio.run_coroutine_threadsafe() to dispatch coroutines to the Python event loop.

    The required lifecycle flow:

    1. asyncio.run(main()) creates and starts the event loop.
    2. await client.connect() is called inside main(). It uses asyncio.get_running_loop() to store a reference to the active loop.
    3. Go callbacks use that stored loop to schedule your event handlers.
    4. await client.idle() ensures the loop remains active to receive these events.

    Critical Warning:

    • Do NOT use asyncio.get_event_loop() (deprecated/raises errors in newer Python).
    • Do NOT use asyncio.new_event_loop() (creates an orphan loop that will silently fail to execute events unless manually managed).
  4. Use the Neonize Event System

    master

    Neonize uses a decorator-based event system that is type-safe. You can register handlers for specific event types using the @client.event(EventType) decorator. The system supports both synchronous clients (NewClient) and asynchronous clients (NewAClient).

    Common event types include MessageEv for incoming messages and ReceiptEv for message receipts.

    # Synchronous event handling
    @client.event(MessageEv)
    def on_message(client: NewClient, event: MessageEv):
        handle_message(event)
    
    @client.event(ReceiptEv)
    def on_receipt(client: NewClient, event: ReceiptEv):
        handle_receipt(event)
    
    # Asynchronous event handling
    @async_client.event(MessageEv)
    async def on_message(client: NewAClient, event: MessageEv):
        await handle_message_async(event)
  5. Handle different message types in MessageEv

    master

    When processing a MessageEv, you can inspect the event.Message object to determine the type of content received. Common fields include:

    • Text: msg.conversation
    • Image: msg.imageMessage.caption
    • Video: msg.videoMessage.caption
    • Document: msg.documentMessage.fileName
    @client.event(MessageEv)
    def on_message(client: NewClient, event: MessageEv):
        msg = event.Message
        
        # Text message
        if msg.conversation:
            print(f"Text: {msg.conversation}")
        
        # Image with caption
        elif msg.imageMessage:
            print(f"Image: {msg.imageMessage.caption}")
        
        # Video with caption
        elif msg.videoMessage:
            print(f"Video: {msg.videoMessage.caption}")
        
        # Document
        elif msg.documentMessage:
            print(f"Document: {msg.documentMessage.fileName}")
  6. Compare Async vs Sync clients

    master

    Choose the client type based on your use case:

    FeatureAsync ClientSync Client
    Entry pointasyncio.run()client.connect()
    Event handlersasync defdef
    Performance⚡ Higher🐌 Lower
    Concurrency✅ Excellent⚠️ Limited
    Complexity🔴 Higher🟢 Lower
    Best ForProductionPrototyping

    Use the Async Client when you need:

    • High concurrency: Handling many connections simultaneously.
    • Non-blocking I/O: Avoiding blocking on network operations.
    • Integration: Working with async frameworks like FastAPI or aiohttp.
    • Performance: Achieving maximum throughput.
  7. Manage sessions and databases

    master

    Neonize stores session data in a database to allow reconnection without re-authentication. You can specify the database location or type during NewClient initialization.

    • SQLite (Default): Pass a file path to the database argument.
    • PostgreSQL: Pass a connection string to the database argument.
    • Multiple Accounts: Create separate NewClient instances with unique database files to manage multiple WhatsApp accounts simultaneously.
    • Logout: Use client.logout() to clear the current session and force re-authentication on the next connection.
    # SQLite session
    client = NewClient("my_bot", database="./sessions/my_bot.db")
    
    # PostgreSQL session
    client = NewClient(
        "my_bot",
        database="postgresql://user:pass@localhost/whatsapp"
    )
    
    # Logout to clear session
    client.logout()
  8. Register event handlers for WhatsApp events

    master

    Neonize uses an event-driven system. You can register handlers using the @client.event(EventClass) decorator. The handler function must accept (client, event) as arguments.

    Supported event types include:

    • MessageEv: New message received
    • ReceiptEv: Message receipt (read/delivered)
    • PresenceEv: User online/offline status change
    • GroupInfoEv: Group information updates
    • PictureEv: Profile picture changes
    • ConnectedEv: Client connection status
    • PairStatusEv: Pairing status changes
    • LoggedInEv: Successful login
  9. Authenticate multiple WhatsApp accounts

    master

    To use multiple WhatsApp accounts simultaneously, create separate NewClient instances, each with a unique session name. Sessions persist indefinitely unless you logout or WhatsApp revokes them.

    bot1 = NewClient("account1")
    bot2 = NewClient("account2")
  10. Basic usage of the Async Client

    master

    The NewAClient provides full async/await support for high-performance applications. To use it, instantiate NewAClient, define asynchronous event handlers using the @client.event() decorator, and use asyncio.run() as the entry point to manage the event loop.

    Key lifecycle methods:

    • await client.connect(): Establishes the connection and captures the running event loop.
    • await client.idle(): Keeps the client alive and the loop running.
    • await client.stop(): Gracefully stops the client.
    import asyncio
    from neonize.aioze.client import NewAClient
    from neonize.aioze.events import MessageEv, ConnectedEv
    
    client = NewAClient("async_bot")
    
    @client.event(ConnectedEv)
    async def on_connected(client: NewAClient, event: ConnectedEv):
        print("✅ Connected!")
    
    @client.event(MessageEv)
    async def on_message(client: NewAClient, event: MessageEv):
        text = event.Message.conversation
        if text == "ping":
            await client.reply_message("pong!", event)
    
    async def main():
        await client.connect()   # captures the running event loop internally
        await client.idle()       # keeps the client alive
    
    asyncio.run(main())          # ← standard entry point