Recent Releases of wmongo

wmongo - v0.1.0

wmongo

wmongo is a library for python control of MongoDB databases.

Description

wmongo simplifies the use of MongoDB databases in Python. It provides a simple interface to the MongoDB database, allowing you to easily insert, update, and delete data.

Installation

To install the library, use pip:

bash pip install wmongo

Description

The wmongo library offers a number of general-purpose modules.

License

MIT

This project is licensed under the MIT License. See the LICENSE file for details.

Examples

This directory contains a collection of examples that demonstrate the usage of various modules and functionalities in this project. Each subfolder corresponds to a specific module and includes example scripts to help you understand how to use that module.

Directory Structure

The examples are organized as follows:

examples/ wmongo_async/ crud.py notification_receiver.py wmongo/ crud.py delete_and_notify.py insert_and_notify.py notifaction_receiver.py update_and_notify.py

How to Use

Navigate to the module folder of interest, e.g., examples/module1/.
Open the README.md in that folder to get detailed information about the examples.
Run the scripts directly using: bash python example1.py

Modules and Examples

wmongo

Description

This module demonstrates specific functionalities.

crud.py: Example demonstrating functionality.

```python from typing import List from wmongo import WMongo from pydantic import BaseModel

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "uri": "mongodb://localhost:27017", "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

credentials = { *mongo_credentials, *redis_credentials, }

Pydantic model for user data validation

class UserModel(BaseModel): name: str age: int

Insert a user synchronously

userdata = UserModel(name="Bob", age=25).dict() with WMongo(database="mydb", verbose=False, **credentials) as wm: if wm.haspermission(user_id="123", collection="users"): wm.insert("users", {"name": "Alice", "age": 30}) else: print("Access Denied")

Insert a user synchronously with Redis caching

with WMongo(database="mydb", verbose=False, **credentials) as wm: wm.update("users", {"name": "Bob"}, {"age": 26})

Read, update, and delete operations within the context manager

with WMongo(database="mydb", verbose=False, **credentials) as wm: # Delete user wm.delete("users", {"name": "Bob"})

Pydantic model for user data validation

class UserRole(BaseModel): user_id: str role: int collections: List[str] ```

deleteandnotify.py: Example demonstrating functionality.

```python from wmongo import WMongo from pydantic import BaseModel

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

notifications = { "enable_notifications": True, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

Pydantic model for user data validation

class UserModel(BaseModel): name: str age: int

Insert a user synchronously

userdata = UserModel(name="Bob", age=25).dict() with WMongo(database="mydb", verbose=True, **credentials) as wm: if wm.haspermission(user_id="123", collection="users"): wm.delete("users", {"name": "Alice"}) else: print("Access Denied") ```

insertandnotify.py: Example demonstrating functionality.

```python from wmongo import WMongo from pydantic import BaseModel

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

notifications = { "enable_notifications": True, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

Pydantic model for user data validation

class UserModel(BaseModel): name: str age: int

Insert a user synchronously

notifaction_receiver.py: Example demonstrating functionality.

```python from wmongo import WMongo

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

def mynotificationcallback(message: str): print(f"📩 Notification Received: {message}")

notifications = { "enablenotificationreceiver": True, "notificationcallback": mynotification_callback, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

Este script solo escuchará notificaciones

with WMongo( database="mydb", verbose=False, **credentials, ) as wm: print("Listening for notifications...") wm.listen_notifications() ```

updateandnotify.py: Example demonstrating functionality.

```python from wmongo import WMongo from pydantic import BaseModel

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

notifications = { "enable_notifications": True, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

Pydantic model for user data validation

class UserModel(BaseModel): name: str age: int

Insert a user synchronously

userdata = UserModel(name="Bob", age=25).dict() with WMongo(database="mydb", verbose=True, **credentials) as wm: if wm.haspermission(user_id="123", collection="users"): wm.update("users", {"name": "Alice"}, {"age": 31}) else: print("Access Denied") ```

wmongo_async

Description

This module demonstrates specific functionalities.

crud.py: Example demonstrating functionality.

```python from wmongo import WMongoAsync import asyncio

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

notifications = { "enable_notifications": True, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

async def insertexample(): async with WMongoAsync(database="mydb", verbose=False, **credentials) as wm: docid = await wm.insert("users", {"name": "Alice", "age": 30}) print(f"Inserted document with ID: {doc_id}")

asyncio.run(insert_example())

async def find_example(): async with WMongoAsync(database="mydb", verbose=False, **credentials) as wm: users = await wm.find("users", {"age": {"$gte": 20}}) # Usuarios con edad >= 20 print(f"Users found: {users}")

asyncio.run(find_example())

async def updateexample(): async with WMongoAsync(database="mydb", verbose=False, **credentials) as wm: updatedcount = await wm.update("users", {"name": "Alice"}, {"age": 35}) print(f"Updated {updated_count} documents.")

asyncio.run(update_example())

async def deleteexample(): async with WMongoAsync(database="mydb", verbose=False, **credentials) as wm: deletedcount = await wm.delete("users", {"name": "Alice"}) print(f"Deleted {deleted_count} documents.")

asyncio.run(delete_example()) ```

notification_receiver.py: Example demonstrating functionality.

```python from wmongo import WMongoAsync import asyncio

Credentials stored in a dictionary for easy reuse

mongo_credentials = { "username": "root", "password": "example", }

rediscredentials = { "redishost": "localhost", "redisport": 6379, "redisdb": 0, }

def mynotificationcallback(message: str): print(f"📩 Notification Received: {message}")

notifications = { "enablenotificationreceiver": True, "notificationcallback": mynotification_callback, }

credentials = { *mongo_credentials, *redis_credentials, **notifications, }

async def listenfornotifications(): async with WMongoAsync(database="mydb", verbose=False, **credentials) as wm: print("Listening for notifications...") wm.listen_notifications()

asyncio.run(listenfornotifications()) ```

- Python
Published by wisrovi over 1 year ago