assignment1

handler/__init__.py

+
+

main.py

+from fastapi import FastAPI
+from service import read_violation,read_noviolation,Fnd,check_result
+
+app = FastAPI()
+
+
+@app.get("/getdata1")
+def most_violation():
+    return read_violation()
+
+@app.get("/getdata2")
+def no_violation():
+    return read_noviolation()
+
+@app.get('/getdata')
+def get_data():
+    return Fnd()
+
+@app.get('/getDetails')
+def checkdata(business_name ,record_type):
+    return check_result(business_name ,record_type)
\ No newline at end of file

pydantic_1/__init__.py

+from pydantic import Basemodel
+
+class Employee_details(BaseModel):
+
+    id: str
+
+    certificate_number: int
+
+    business_name: str
+
+    date: str
+
+    result: str
+
+    sector: str
+
+    address: dict
+

pydantic_1/model.py

+import pandas as pd
+
+import pymongo
+
+from pydantic_1 import BaseModel
+
+client = pymongo.MongoClient("mongodb://192.168.0.220:2747/")
+
+mydb = client["factory"]
+
+mycol = mydb["status"]
+
+
+# cnt = mycol.count_documents({"result" : "Violation Issued"})
+
+# print(cnt,"companies has violation issued")
+
+
+#
+# class Employee_details(BaseModel):
+#
+#     id: str
+#
+#     certificate_number: int
+#
+#     business_name: str
+#
+#     date: str
+#
+#     result: str
+#
+#     sector: str
+#
+#     address: dict
+#
+
+
+def Fnd():
+    q = mycol.find({})
+
+    if q:
+
+        a = []
+
+        for i in q:
+            d = {'Business Name': i['business_name'], 'Date': i["date"], 'Result': i["result"]}
+
+            a.append(d)
+
+        a = pd.DataFrame(a)
+
+        return (a.to_excel("incpection.xlsx", sheet_name="mySheet", index=False))
+
+    else:
+
+        return "no data"
+
+
+def Insert(user):
+    mycol.insert_one(user.dict())
+
+    return "Inserted"

service/__init__.py

+import pymongo
+import pandas as pd
+from pydantic import BaseModel
+
+myclient = pymongo.MongoClient("mongodb://192.168.0.220:2747/")
+mydb = myclient["factory"]
+mycol = mydb["status"]
+
+class Employee_details(BaseModel):
+
+    id: str
+    certificate_number: int
+    business_name: str
+    date: str
+    result: str
+    sector: str
+    address: dict
+#
+
+dblist = myclient.list_database_names()
+if "factory" in dblist:
+    print("The database exists.")
+
+
+mycol.find()
+
+
+def read_violation():
+    agg = list(mycol.aggregate(
+        [
+            {
+                '$project': {
+                    '_id': '$business_name',
+                    'result': 'Violation Issued'
+                }
+            }, {
+            '$group': {
+                '_id': '$_id',
+                'heghest': {
+                    '$max': '$result'
+                }
+            }
+        }, {
+            '$sort': {
+                'highest': -1
+            }
+        }
+        ]))
+    return (agg[0])
+
+
+def read_noviolation():
+    r = mycol.find({"result": "No Violation Issued"})
+    lst = []
+    for i in r:
+        lst.append(i["business_name"])
+    return lst
+
+
+
+
+
+
+
+
+class Employee_details(BaseModel):
+
+    id: str
+    certificate_number: int
+    business_name: str
+    date: str
+    result: str
+    sector: str
+    address: dict
+
+
+
+def Fnd():
+    q = mycol.find({})
+    if q:
+        a = []
+        for i in q:
+            d = {'Business Name': i['business_name'], 'Date': i["date"], 'Result': i["result"]}
+            a.append(d)
+        a = pd.DataFrame(a)
+        return (a.to_excel("incpection.xlsx", sheet_name="mySheet", index=False))
+    else:
+        return "no data"
+
+
+def Insert(user):
+    mycol.insert_one(user.dict())
+    return "Inserted"
+
+def check_result(business_name, record_type):
+    x = business_name
+    y = record_type
+    cnt = mycol.count_documents({'$and':[{'business_name':x},{'result':y}]})
+    return str(cnt) +" " +str( y)
\ No newline at end of file