Commit b68cc1af authored by ramya.r's avatar ramya.r

first

parent c88ecedd
[SERVICE]
port=7988
host=0.0.0.0
[MONGO-DB]
mongo_uri=mongodb://192.168.0.220:2717/
import uvicorn
from fastapi import FastAPI
from scripts.core.engine.mconnection import get_database
from script.config.app_config import Service
from script.service.employee import router
app = FastAPI(title="Employee")
app.include_router(router)
if __name__ == "__main__":
get_database()
uvicorn.run("scripts.services.api_call:app")
\ No newline at end of file
uvicorn.run("main:app", host=Service.host, port=Service.port)
from configparser import SafeConfigParser
config = SafeConfigParser()
config.read('conf/application.conf')
class Service:
port = int(config.getint("SERVICE", "port"))
host = config.get("SERVICE", "host")
class Mongo:
mon = config.get('MONGO-DB', 'mongo_uri')
class APIEndpoints:
delete = "/delete"
save = "/save"
employee_base = "/employee"
insert = "/insert"
update = "/update"
find = "/find"
\ No newline at end of file
database_name = "interns2023"
\ No newline at end of file
from script.core.db.mongo.interns2023 import database_name
class EmployeeDetails:
def __init__(self, mongo_client):
self.mongo_client = mongo_client
self.database_name = database_name
self.collection_name = "employee"
def insert_item(self, data):
self.insert_one(data)
def list_items(self, data):
self.find_one(data)
def delete_item(self, data):
self.delete_one(data)
def update_item(self, data):
self.update_one(data)
class NameDoesNotExist(Exception):
pass
\ No newline at end of file
from script.core.db.mongo.interns2023.employee import EmployeeDetails
from script.core.errors import NameDoesNotExist
class EmployeeData:
def __init__(self):
self.employee_col = EmployeeDetails(mongo_client="")
@staticmethod
def insert_data(self,emp_id: int):
try:
d = {"emp_id": 101}
if 101 not in d.values():
raise NameDoesNotExist
except Exception as e:
print(e, "Error Detected in insertion")
@staticmethod
def delete_data(self, emp_id: int):
try:
d = {"emp_id": 103}
if 103 not in d.values():
raise NameDoesNotExist
except ValueError:
raise ValueError("name")
except Exception as e:
print(e, "Error Detected in Deletion")
@staticmethod
def update_data(self, emp_id: int):
try:
d = {"emp_id": 102}
if 102 not in d.values():
raise NameDoesNotExist
except Exception as e:
print(e, "Error Detected in Updation")
@staticmethod
def find_data(self, emp_id: int):
try:
d = {"emp_id": 105}
if 105 not in d.values():
raise NameDoesNotExist
except Exception as e:
print(e, "Error Detection in Finding")
from typing import Optional
from pydantic import BaseModel
# define a model for Item
class Employee(BaseModel):
emp_id: Optional[int]
emp_name: str
emp_department: str
emp_sal: int
from typing import Any, Optional
from pydantic import BaseModel
class DefaultResponse(BaseModel):
status: str = "failed"
message: str
data: Optional[Any]
import logging
from fastapi.routing import APIRouter
from script.constants import APIEndpoints
from script.core.handlers.employee import EmployeeData
from script.core.schema.response import DefaultResponse
router = APIRouter(prefix=APIEndpoints.employee_base)
handler = EmployeeData()
@router.post(APIEndpoints.delete)
async def delete_item(emp_id: int):
"""This API deletes an item"""
try:
handler.delete_data(emp_id,int)
return DefaultResponse(message="Successfully deleted", status="success")
except ValueError:
return DefaultResponse(message="Due to Value Error")
except Exception as e:
logging.exception(e)
return DefaultResponse(message="Deletion failed due to server error")
@router.post(APIEndpoints.update)
async def update_item(emp_id: int):
try:
handler.update_data(emp_id,int)
return DefaultResponse(message="Successfully Updated", status="success")
except ValueError:
return DefaultResponse(message="Due to value error")
except Exception as e:
logging.exception(e)
return DefaultResponse(message="Updation Failed due to server error")
@router.get(APIEndpoints.find)
async def find_item(emp_id: int):
try:
handler.find_data(emp_id,int)
return DefaultResponse(message="Successfully Found", status="success")
except ValueError:
return DefaultResponse(message="Due to value error")
except Exception as e:
logging.exception(e)
return DefaultResponse(message="Finding Failed due to server error")
@router.get(APIEndpoints.insert)
async def insert_item(emp_id: int):
try:
handler.insert_data(emp_id,int)
return DefaultResponse(message="Successfully Found", status="success")
except ValueError:
return DefaultResponse(message="Due to value error")
except Exception as e:
logging.exception(e)
return DefaultResponse(message="Finding Failed due to server error")
from fastapi import APIRouter
from pymongo import MongoClient
router = APIRouter()
# Create a MongoClient instance
client = MongoClient("Mongo_uri")
# Connect to a database
db = client.mydatabase
from typing import Dict, List, Optional
from pymongo import MongoClient
import logging
class MongoConnect:
def __init__(self, uri):
try:
self.uri = uri
self.client = MongoClient(self.uri, connect=False)
except Exception as e:
logging.error(f"Exception in connection {(str(e))}")
raise e
class MongoCollectionBaseClass:
def __init__(self, mongo_client, database, collection):
self.client = mongo_client
self.database = database
self.collection = collection
def insert_one(self, data: Dict):
try:
database_name = self.database
collection_name = self.collection
db = self.client[database_name]
collection = db[collection_name]
response = collection.insert_one(data)
return response.inserted_id
except Exception as e:
logging.error(f"Error in inserting the data {str(e)}")
raise e
def find_one(self, query: Dict, filter_dict: Optional[Dict] = None):
try:
database_name = self.database
collection_name = self.collection
if filter_dict is None:
filter_dict = {"_id": 0}
db = self.client[database_name]
collection = db[collection_name]
response = collection.find_one(query, filter_dict)
return response
except Exception as e:
logging.error(f"Failed to fetch {str(e)}")
raise e
def update_one(self, query: Dict, data: Dict, upsert: bool = False, ):
try:
database_name = self.database
collection_name = self.collection
db = self.client[database_name]
collection = db[collection_name]
response = collection.update_one(query, {"$set": data}, upsert=upsert)
return response.modified_count
except Exception as e:
logging.error(f"Failed to update one doc {str(e)}")
raise e
def delete_one(self, query: Dict):
try:
database_name = self.database
collection_name = self.collection
db = self.client[database_name]
collection = db[collection_name]
response = collection.delete_one(query)
return response.deleted_count
except Exception as e:
logging.error(f"Failed to delete {str(e)}")
raise e
from pymongo import MongoClient
def get_database():
uri = "mongodb://intern_23:intern%40123@192.168.0.220:2717/?serverSelectionTimeoutMS=5000&connectTimeoutMS=10000&authSource=interns_b2_23&authMechanism=SCRAM-SHA-256"
client = MongoClient(uri)
return client['interns_b2_23']
dbname = get_database()
employee = dbname["employee"]
\ No newline at end of file
from scripts.core.engine.mconnection import employee
employee_5= {
"emp_id" : "105",
"emp_name" : "Aswin",
"emp_department" : "Data scientist",
"emp_sal" : 60000
}
def insert_data():
employee.insert_one(employee_5)
return employee_5
def update_data():
employee.update_one({"emp_name": "Alex"}, {"$set": {"emp_name": "jerry"}})
def delete_data():
employee.delete_one({"emp_name": "Aruna"})
def find_data():
employee.employeeDB.find({"employee_name": "Meera"})
from fastapi import FastAPI
from scripts.core.handlers.crud import insert_data, update_data, delete_data, find_data
app = FastAPI()
@app.get("/insertion")
async def insert():
data = insert_data()
return "inserted"
@app.get("/updation")
async def update():
data = update_data()
return "updated"
@app.get("/deletion")
async def delete():
data = delete_data()
return "deleted"
@app.get("/find")
async def find():
data = find_data()
return "founded"
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment