Effortless Data Storage: MongoDB Database and Raspberry Pi Pico W Walkthrough - Part 2 (CRUD)

Learn how to use basic CRUD operations with MongoDB and the Raspberry Pi Pico W using their Data API. Before getting started make sure you watch Part 1. In this tutorial, we use the BME280 sensor to get temperature, sensor, and humidity data to work within our database. In reality, you can use any sensor you like!

You can purchase the BME280 from here:

BME280 Pre-Soldered

Also, be sure to subscribe and support the channel if you have not!

Subscribe:

Youtube

Support:

https://www.buymeacoffee.com/mmshilleh

Step 1-) Code

from machine import Pin, I2C, RTC
import json
import urequests as requests
import network
import ntptime
import time
import utime

import bme280

import constants


i2c = I2C(0,sda=Pin(0), scl=Pin(1), freq=400000)
URL = "<url>"
API_KEY = "<api key>"

def connect_to_wifi(ssid, psk):
    wlan = network.WLAN(network.STA_IF)
    wlan.active(True)
    wlan.connect(ssid, psk)

    while not wlan.isconnected() and wlan.status() >= 0:
        print("Waiting to Connect")
        time.sleep(10)
    if not wlan.isconnected():
        raise Exception("Wifi not available")
    print("Connected to WiFi")


def findOne(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "findOne", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def find(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "find", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def insertOne(temp, pressure, humidity, time):
    try:
        headers = { "api-key": API_KEY }
        documentToAdd = {"Device": "BME280",
                         "Temperature (C)": temp,
                         "Pressure": pressure,
                         "Humidity": humidity,
                         "Time": time}
        insertPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "document": documentToAdd,
        }
        response = requests.post(URL + "insertOne", headers=headers, json=insertPayload)
        print(response)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def insertMany(document_list):
    try:
        headers = { "api-key": API_KEY }
        insertPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "documents": document_list,
        }
        response = requests.post(URL + "insertMany", headers=headers, json=insertPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def updateOne(filter_dictionary, update_dict):
    try:
        headers = { "api-key": API_KEY }
        update = {"set": update_dict}
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
            "update": update_dict,
        }
        response = requests.post(URL + "updateOne", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)
        
        
def deleteOne(filter_dictionary):
    try:
        headers = { "api-key": API_KEY }
        searchPayload = {
            "dataSource": "Cluster0",
            "database": "BME280",
            "collection": "Readings",
            "filter": filter_dictionary,
        }
        response = requests.post(URL + "delete", headers=headers, json=searchPayload)
        print("Response: (" + str(response.status_code) + "), msg = " + str(response.text))
        if response.status_code >= 200 and response.status_code < 300:
            print("Success Response")
        else:
            print(response.status_code)
            print("Error")
        response.close()
    except Exception as e:
        print(e)


def main():
    connect_to_wifi(constants.INTERNET_NAME, constants.INTERNET_PASSWORD)
    #document_list = []
    while True:
        bme = bme280.BME280(i2c=i2c)
        temp, pressure, humidity = bme.values
        print(temp, pressure, humidity)
        rtc_time_tuple = RTC().datetime()
        formatted_time = "{:04}-{:02}-{:02} {:02}:{:02}:{:02}".format(
            rtc_time_tuple[0], rtc_time_tuple[1], rtc_time_tuple[2], 
            rtc_time_tuple[4], rtc_time_tuple[5], rtc_time_tuple[6]
        )
        print(formatted_time)
        
        #insertOne(temp, pressure, humidity, formatted_time)
        
        #document_list.append({"Device": "BME280",
            #"Temperature (C)": temp,
            #"Pressure": pressure,
            #"Humidity": humidity,
            #"Time": formatted_time}
        #)
        #if len(document_list) == 10:
           #print(json.dumps(document_list))
           #insertMany(document_list)
           #document_list = []
        #findOne({"Temperature (C)": "23.26C", "Humidity": "53.69%"})
        #find({"Temperature (C)": "24.65C"})
        #updateOne({"Temperature (C)": "23.26C"}, {"Temperature (C)": "24.26C"})
        deleteOne({"Temperature (C)": "24.26C"})

main()

This is the code we use in this tutorial, we comment and uncomment parts of the main function to activate different functions; this is more clear in the Youtube Video. Also, note that the code can be cleaned up considerably, there is a lot of repetition, mainly for demonstration purposes.

    Overall the operations with their API are very similar, just a difference in URL extension along with slightly different values in the payload. They have much more information in their documentation here. Nonetheless, this should give you the basis to do most things you have to do when using MongoDB with Data API.

    Conclusion

    Hope you enjoyed the tutorial and the series thus far. MongoDB makes it incredibly easy to use their cloud storage for free! This can be incredibly useful for cases where you want to store and access slow-moving data. Do not forget to subscribe to the channel above and stay tuned for more coding and microcontroller-related content.

    Create a free account to access full content.

    All access to code and resources on ShillehTek.

    Signup Now

    Already a member? Sign In

    Explore More on Our Blog

    How to Use ADS1115 with the Raspberry Pi (Part 1)

    How to Use ADS1115 with the Raspberry Pi (Part 1)

    Discover how to expand your Raspberry Pi projects by integrating the ADS1115 ADC for precise analog signal reading....

    How to Install Pip Packages in AWS Lambda Using Docker and ECR

    How to Install Pip Packages in AWS Lambda Using Docker and ECR

    Learn how to streamline AWS Lambda deployments by using Docker and Amazon Elastic Container Registry (ECR) to package...

    Create Tabular Product Descriptions on Your Shopify Store

    Create Tabular Product Descriptions on Your Shopify Store

    Enhance your Shopify store's product pages with our comprehensive guide on implementing tabular descriptions. Learn how to add a...

    SSH Into Raspberry Pi with Tailscale VPN

    SSH Into Raspberry Pi with Tailscale VPN

    Effortlessly access and manage your Raspberry Pi from anywhere using Tailscale's secure mesh VPN.

    Send Email with Lua and the ESP32

    Send Email with Lua and the ESP32

    In this tutorial, we delve into sending emails with the ESP32-S3 using Lua, focusing on the Xedge IDE's built-in SMTP...

    How to Code with Lua on ESP32 with XEdge32

    How to Code with Lua on ESP32 with XEdge32

    Learn how to set up Xedge32 and start coding on the ESP32-S3 with Lua programming!

    Stream Audio From Raspberry Pi to Local Computer

    Stream Audio From Raspberry Pi to Local Computer

    Discover the simplicity of streaming live audio directly from a USB microphone connected to your Raspberry Pi to...

    SSH Raspberry Pi via Cell Phone

    SSH Raspberry Pi via Cell Phone

    This beginner-friendly guide will walk you through remotely controlling your Raspberry Pi using SSH through your cell phone.

    Remotely Control Raspberry Pi via SSH from External Network

    Remotely Control Raspberry Pi via SSH from External Network

    Learn how to SSH into your Raspberry Pi from any network. This is critical in IoT since you can control...

    Stream Video from Raspberry Pi Camera to YouTube Live

    Stream Video from Raspberry Pi Camera to YouTube Live

    Learn how to stream to YouTube from a Raspberry Pi Camera.

    Back to blog

    Leave a comment

    Please note, comments need to be approved before they are published.