2
\$\begingroup\$

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 :

  1. Route to get all objects from a collection.
  2. Route to get one object from a collection by a specific identifier (doc_id).
  3. Route to post a new object to a collection.
  4. 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
        }
\$\endgroup\$

0

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Browse other questions tagged or ask your own question.