Esercizi Lettura MongoDB Python

Codegrind Team•Jul 10 2024

Esercizi sulla lettura di dati da un database MongoDB utilizzando le librerie Python pymongo e motor. Gli esercizi includono approcci sequenziali e orientati agli oggetti (OOP).

Esercizio 1: Lettura Singola con pymongo (Sequenziale)

Leggere un singolo documento da una collezione utilizzando pymongo.

from pymongo import MongoClient

def read_single_document():
    client = MongoClient('mongodb://localhost:27017/')
    db = client['testdb']
    collection = db['employees']
    document = collection.find_one({"name": "John Doe"})
    print("Documento trovato:", document)
    client.close()

read_single_document()

Esercizio 2: Lettura Multipla con pymongo (OOP)

Leggere più documenti da una collezione utilizzando pymongo e OOP.

from pymongo import MongoClient

class MongoDBConnection:
    def __init__(self, uri, db_name, collection_name):
        self.client = MongoClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    def read_multiple_documents(self, query):
        documents = self.collection.find(query)
        for document in documents:
            print("Documento trovato:", document)

    def close(self):
        self.client.close()

# Uso
db_connection = MongoDBConnection('mongodb://localhost:27017/', 'testdb', 'employees')
db_connection.read_multiple_documents({})
db_connection.close()

Esercizio 3: Lettura Singola con motor (Sequenziale)

Leggere un singolo documento da una collezione utilizzando motor.

import motor.motor_asyncio
import asyncio

async def read_single_document():
    client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['testdb']
    collection = db['employees']
    document = await collection.find_one({"name": "Jane Doe"})
    print("Documento trovato:", document)
    client.close()

# Uso
asyncio.run(read_single_document())

Esercizio 4: Lettura Multipla con motor (OOP)

Leggere più documenti da una collezione utilizzando motor e OOP.

import motor.motor_asyncio
import asyncio

class MongoDBConnection:
    def __init__(self, uri, db_name, collection_name):
        self.client = motor.motor_asyncio.AsyncIOMotorClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    async def read_multiple_documents(self, query):
        async for document in self.collection.find(query):
            print("Documento trovato:", document)

    def close(self):
        self.client.close()

# Uso
async def main():
    db_connection = MongoDBConnection('mongodb://localhost:27017/', 'testdb', 'employees')
    await db_connection.read_multiple_documents({})
    db_connection.close()

asyncio.run(main())

Esercizio 5: Lettura Condizionata con pymongo (Sequenziale)

Leggere documenti con una condizione specifica utilizzando pymongo.

from pymongo import MongoClient

def read_conditional_documents():
    client = MongoClient('mongodb://localhost:27017/')
    db = client['testdb']
    collection = db['employees']
    documents = collection.find({"position": "Data Scientist"})
    for document in documents:
        print("Documento trovato:", document)
    client.close()

read_conditional_documents()

Esercizio 6: Lettura con Ordinamento con pymongo (OOP)

Leggere documenti da una collezione con ordinamento utilizzando pymongo e OOP.

from pymongo import MongoClient

class MongoDBConnection:
    def __init__(self, uri, db_name, collection_name):
        self.client = MongoClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    def read_sorted_documents(self, query, sort_field):
        documents = self.collection.find(query).sort(sort_field)
        for document in documents:
            print("Documento trovato:", document)

    def close(self):
        self.client.close()

# Uso
db_connection = MongoDBConnection('mongodb://localhost:27017/', 'testdb', 'employees')
db_connection.read_sorted_documents({}, "name")
db_connection.close()

Esercizio 7: Lettura con Limite con motor (Sequenziale)

Leggere un numero limitato di documenti da una collezione utilizzando motor.

import motor.motor_asyncio
import asyncio

async def read_limited_documents():
    client = motor.motor_asyncio.AsyncIOMotorClient('mongodb://localhost:27017/')
    db = client['testdb']
    collection = db['employees']
    cursor = collection.find({}).limit(2)
    async for document in cursor:
        print("Documento trovato:", document)
    client.close()

# Uso
asyncio.run(read_limited_documents())

Esercizio 8: Lettura Condizionata con motor (OOP)

Leggere documenti con una condizione specifica utilizzando motor e OOP.

import motor.motor_asyncio
import asyncio

class MongoDBConnection:
    def __init__(self, uri, db_name, collection_name):
        self.client = motor.motor_asyncio.AsyncIOMotorClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    async def read_conditional_documents(self, query):
        async for document in self.collection.find(query):
            print("Documento trovato:", document)

    def close(self):
        self.client.close()

# Uso
async def main():
    db_connection = MongoDBConnection('mongodb://localhost:27017/', 'testdb', 'employees')
    await db_connection.read_conditional_documents({"position": "Developer"})
    db_connection.close()

asyncio.run(main())

Esercizio 9: Lettura con Filtro di Campo con pymongo (Sequenziale)

Leggere documenti selezionando specifici campi utilizzando pymongo.

from pymongo import MongoClient

def read_field_filtered_documents():
    client = MongoClient('mongodb://localhost:27017/')
    db = client['testdb']
    collection = db['employees']
    documents = collection.find({}, {"name": 1, "position": 1, "_id": 0})
    for document in documents:
        print("Documento trovato:", document)
    client.close()

read_field_filtered_documents()

Esercizio 10: Lettura con Proiezione di Campo con motor (OOP)

Leggere documenti selezionando specifici campi utilizzando motor e OOP.

import motor.motor_asyncio
import asyncio

class MongoDBConnection:
    def __init__(self, uri, db_name, collection_name):
        self.client = motor.motor_asyncio.AsyncIOMotorClient(uri)
        self.db = self.client[db_name]
        self.collection = self.db[collection_name]

    async def read_field_projection_documents(self, query, projection):
        async for document in self.collection.find(query, projection):
            print("Documento trovato:", document)

    def close(self):
        self.client.close()

# Uso
async def main():
    db_connection = MongoDBConnection('mongodb://localhost:27017/', 'testdb', 'employees')
    await db_connection.read_field_projection_documents({}, {"name": 1, "position": 1, "_id": 0

})
    db_connection.close()

asyncio.run(main())