The project’s purpose to demonstrate how to work with some of the crud functionality within MongoDB in relation with the Python programming language. In this particular initial project, we will only be using the first to letters of CRUD functions which to (CREATE) and (READ). Our testing will be done in Jupyter-Notebook which will showcase if our logic from our python script is functioning correctly and that our server is running within our python object AnimalShelter.
Let me brake down what tools I will be using. My IDE is DataSpell JetBrains which is an IDE specifically made for data science and jupyter-notebook integration right in the IDE. Which PyCharm does not. I began the project by creating a simple python project in a conda environment (Virtual Environment). With two files inside. A animal_shelter.py file and animals_db_script.ipynb file. I used the boilerplate code provided in the rubrics for this assignment and added or refactored the code to fit more of how I wanted to represent the AnimalShelter object. I am huge on using all programming languages with types and I think since python 3.7, you can implement types in python that though they do not work at compile time, do work at runtime. Which can make your python code much faster. I also added some utility functions.
- An example of my custom types:
# ========== custom utility types ==========
class AnimalData(TypedDict):
age_upon_outcome: str
animal_id: str
animal_type: str
breed: str
color: str
date_of_birth: str
datetime: str
monthyear: str
name: str
outcome_subtype: str
outcome_type: str
sex_upon_outcome: str
location_lat: int
location_long: int
age_upon_outcome_in_weeks: int
class SearchType(TypedDict):
animal_id: str
# ========== utility methods ==========
def pretty_print_json(response):
formatted_response = json.dumps(response, indent=2)
return formatted_responseInside of the AnimalShelter object there is a lot of things going on. In our constructor we are using two parameters. Which is the username and password. I are also creating an object instance to our server, which when a new AnimalShelter instance is created, I can pass as arguments our username and password. I am also creating a database property that provides us our database of AAC. The fact that it is available everywhere in the AnimalShelter object is a bit confusing to me, since in other languages it would be scoped to the constructor and would only be available on a new object instance that is created.
def __init__(self, username: str, password: str):
# Initializing the MongoClient. This helps to
# access the MongoDB databases and collections.
self.client = MongoClient(host=f'mongodb://{username}:{password}@localhost:27017/AAC')
# where xxxx is your unique port number
self.database = self.client['AAC']The create method in the AnimalShelter object was mostly complete but it required that if a dictionary (json like object with keys & values) is passed as an argument to the method that the method will return True or else False. Which the logic is in the code example below:
# Complete this create method to implement the (CREATE) in CRUD.
def create(self, data: list[AnimalData]): # typed the data with a custom type (TypedDict)
if data is None:
raise Exception("Nothing to save, because data parameter is empty")
inserted = self.database.animals.insert_one(data) # data should be dictionary
if inserted != 0: # Will return True if insert is not == 0
print(f"\n1. (CREATE) newly added animal: True")
return True
else:
print(f"\n1. (CREATE) newly added animal: False")
return FalseThe read method had search parameter that called the find and not the find_one method. Since they work differently. I then serialized the data so that it can be outputted in json format.
# Create method to implement the (READ) in CRUD.
def read(self, search: SearchType): # SearchType = { animal_id: str }<- my search criteria
if search is None:
raise Exception("could not find search result...")
if search:
search_response = self.database.animals.find(search, {'_id': False})
json_output = json.loads(json_util.dumps(search_response))
if not json_output:
print(f" Running..(READ) [No item found]: {json_output}")
else:
print(
f"2. (READ) in CRUD by searching animal_id:"
f" {pretty_print_json(json_output)}"
)
return json_outputI then created the update method which would call the update_one. The set part of the arguments passed was where I would pass my parameter of updated_data that I would pass as an argument when the data would update based on the filtered query. I then would call find for the updated_data, serialize it to be able to output it as json.
# (UPDATE) method in CRUD
def update(self, filtered_query, updated_data):
if filtered_query is None or updated_data is None:
raise Exception("Document failed to update")
# updates the data based on the filtered queried type { 'key': 'value' }
self.database.animals.update_one(filtered_query, {'$set': updated_data})
# searches for the updated data
output = self.database.animals.find(updated_data)
# makes the data serializable
json_output = json.loads(json_util.dumps(output))
# prints the data in formatted json form
print(
f"3. (UPDATE) in CRUD by updating the animal_type:"
f" {pretty_print_json(json_output)}"
)
return json_outputFinally, I created the delete CRUD method. I added a parameter to the delete method that queries the data based on the query. It could be anything within the collection. I then called delete on that query. Called find on the queried data, made it serializable, read the data one more time to make sure it was returning empty->[ ] and then outputting a formatted output. While also returning the json_output.
# (DELETE) method in CRUD
def delete(self, query):
"""Delete document fed from parameter"""
if query is None:
raise Exception("Document failed to delete")
# deletes the data based on the query { 'key': 'value' }
self.database.animals.delete_one(query)
# searches for the queried data
output = self.database.animals.find(query)
# makes the data serializable
json_output = json.loads(json_util.dumps(output))
# runs read again that should verify that the data was deleted by returning ->[]
search_for_deleted_data = self.read({'animal_id': 'A238675'})
# a conditional variable that will return `True` if no data is found
is_data_deleted: bool = search_for_deleted_data == []
# formatted output
# calling read to make sure it deleted
print(
f"4. After (READ) has ran.. Will return `True` if {query}"
f" was deleted: {is_data_deleted}"
)
print(f" Delete Complete for: {query}")
print(f" Should return an empty array: {json_output}")
return json_outputIn terms of testing if my script worked I created a jupyter-notebook file an imported all of the api’s that would need in the test file to test my python code.
from animal_shelter import SearchType, AnimalShelter, AnimalDataI created some dummy data to be able to insert into the database and typed it with my custom AnimalData type, that I created in my python script. I also needed an instance of the AnimalShelter object in the notebook.
# ========== dummy data for testing ==========
DUMMY_DATA: AnimalData = {
'age_upon_outcome': '3 year',
'animal_id': 'A238675',
'animal_type': 'Dog',
'breed': 'German Shepperd',
'color': 'Black',
'date_of_birth': '2022-09-03',
'datetime': '2022-09-03 11:43:00',
'monthyear': '2022-09-0T11:43:00',
'name': 'Jose',
'outcome_subtype': 'Foster',
'outcome_type': 'Adoption',
'sex_upon_outcome': 'Male',
'location_lat': 30.5202877527844,
'location_long': -55.6574204375118,
'age_upon_outcome_in_weeks': 15.0833333333333,
}
# creating an animal shelter object instance
animal_shelter = AnimalShelter(username='aacuser', password='alias111')I then called the create method from my python script and then stored it in a variable created animal. The formatting is done in they script file. So, all I have to do is call the create method and it will print out in the correct format.
""" •••••••••
(CREATE) adding a new animal from dummy data to the animal’s collection
••••••••• """
was_animal_created: bool = animal_shelter.create(DUMMY_DATA)
I then queried the newly added data to the collection by way of a custom search query I created that I created a type for. Which would query based on the animal_id. I stored the call of the read method in a variable by the name of search_response. Which will print out a formatted version.
""" ••••••••• (READ) creating a search criteria to search for the new insert ••••••••• """
search: SearchType = {"animal_id": "A238675"}
search_response: Any = animal_shelter.read(search)
I then tested the update method which takes two parameters as arguments. In this case I am using the filtered_query of the animal_id to find the animal I created with the create method. I updated the animal_type from Dog to Wolf.
""" ••••••••• updating the data inserted ••••••••• """
update_response: Any = animal_shelter.update(
filtered_query={'animal_id': 'A238675'},
updated_data={'animal_type': 'Wolf'}
)
Testing the delete was straight forward. I just passed an argument querying the animal_id and the item was deleted.
""" ••••••••• deleting data ••••••••• """
data_to_delete_response = animal_shelter.delete({'animal_id': 'A238675'})