So I wrote a a RESTfull api for a mongodb with python3.7, fastapi and mongoengine and id love to get feedback on how I should make my code more readable, clean, and dry.
basically there are 4 main routes :
- Route to get all objects from a collection.
- Route to get one object from a collection by a specific identifier (doc_id).
- Route to post a new object to a collection.
- Route to update an object from the db by a specific identifier (doc_id).
#db_api.py from fastapi.middleware.cors import CORSMiddleware from fastapi import FastAPI from typing import Dict from mongoengine import connect,disconnect from mongoengine import errors from schemas import * import os MONGODB_ADDR = os.environ.get("MONGODB_ADDR", "chaos.mongodb.openshift") MONGODB_PORT = os.environ.get("MONGODB_PORT", 8080) DB_NAME = os.environ.get("DB_NAME", "chaos") LISTEN_PORT = int(os.environ.get("LISTEN_PORT", 5001)) app = FastAPI() @app.on_event("startup") async def create_db_client(): try: ## Add authentication connect(db=DB_NAME, host=MONGODB_ADDR, port=MONGODB_PORT) except errors.ServerSelectionTimeoutError: print("error connecting to mongodb") @app.on_event("shutdown") async def shutdown_db_client(): pass @app.get("/{collection}") def read_object(collection: Collections): # Get all documents from collection collection_object = get_collection_object(collection) output = [document.to_dict() for document in collection_object.objects] return output @app.get("/{collection}/{doc_id}") def read_one_object(collection: Collections,doc_id: str): try: # Get specific document from collection depending on route # try to get first document, considering there could not be more then 1 collection_object = get_collection_object(collection) doc_identifier_to_doc_id = {collection_object.get_identifier() : doc_id} # Get all objects that have the doc_id and # return the first one as a dict output = collection_object.objects(**doc_identifier_to_doc_id)[0].to_dict() except (IndentationError, IndexError): output = f"No such document '{doc_id}' in '{collection}' collection" return output @app.post("/{collection}/{doc_id}") def write_object(collection: Collections, document: Dict = {}): try: collection = get_collection_object(collection) document = collection(**document) document.save() return document.to_dict() except TypeError as E: print(E) return 'missing parameters' except (errors.FieldDoesNotExist , errors.ValidationError): return 'unknown parameter ' @app.put("/{collection}/{doc_id}") def update_object(collection: Collections, doc_id : str, document: Dict = {}): try: collection_object = get_collection_object(collection) doc_identifier_to_doc_id = {collection_object.get_identifier() : doc_id} document_object = collection_object.objects(**doc_identifier_to_doc_id)[0] document_object.update(**document) document_object.save() document_object = collection_object.objects(**doc_identifier_to_doc_id)[0] return document_object.to_dict() except TypeError: return 'missing parameters' except (errors.FieldDoesNotExist , errors.ValidationError): return 'unknown parameter ' def get_collection_object(collection): if collection == Collections.servers: document = Server elif collection == Collections.groups: document = Group elif collection == Collections.logs: document = Log return document Ive created schemas for the collections I have in my db (ive already dicided to create a parent abstract class for all my documents that specifies the to_dict function and other stuff all of them need).
#schemas.py from mongoengine import * from enum import Enum class Collections(str, Enum): servers = "servers" groups = "groups" logs = "logs" class Log(Document): meta = {'collection': 'logs'} name = StringField() logs = DictField() successful = BooleanField(default=False) date = StringField() victim = StringField() owner = StringField(required=True) @staticmethod def get_identifier(): return 'name' def to_dict(self): return { "name": self.name, "logs": self.logs, "successful": self.successful, "date" : self.date, "victim" : self.victim, "owner": self.owner } class Server(Document): meta = {'collection': 'servers'} dns = StringField(max_length=200, required=True) active = BooleanField(default=False) groups = ListField(StringField(),default=[]) os_type = StringField(required=True) last_fault = StringField() @staticmethod def get_identifier(): return 'dns' def to_dict(self): return { "dns": self.dns, "active": self.active, "groups": self.groups, "os_type": self.os_type, "last_fault": self.last_fault } class Group(Document): meta = {'collection': 'groups'} name = StringField(max_length=200, required=True) active = BooleanField(default=False) last_fault = StringField() owner = StringField(required=True) @staticmethod def get_identifier(): return 'name' def to_dict(self): return { "name": self.name, "active": self.active, "last_fault": self.last_fault, "owner" : self.owner } 
