main.create_event() - Code Metrics - Retr0100/VirgilAPI - Measure and Improve Code Quality continuously with Scrutinizer

main.create_event() A
last analyzed 2023-09-04 23:03 UTC

↳ Parent: main

Complexity

Conditions

Size

Total Lines	26
Code Lines	10

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	3
eloc	10
nop	3
dl	0
loc	26
rs	9.9
c	0
b	0
f	0

"""
    Main file for the API
"""
import datetime
import json
import secrets
import os
from typing import List
import functools

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from pymongo import MongoClient
from fastapi.responses import JSONResponse

# Take the url for connection to MongoDB
urlMongo = os.getenv('MONGO_URL')

# Init the client of Mongo
client = MongoClient(urlMongo)

db = client.virgilUsers
users_collection = db.users
users_collection.create_index("userId", unique=True)
calendar_collection = db.calendarEvent

app = FastAPI()

# Take the base of setting
@functools.lru_cache(maxsize=None)
def get_cached_setting():
    """
    Get all settings from database and cache it in memory

    Returns:
        json: The default settings
    """
    with open('setting.json', 'r',encoding='utf-8') as file:
        return json.load(file)

# Modifica la chiamata nel codice originale
setting = get_cached_setting()

# User Model
class User(BaseModel):
    """
    User model used in this API

    Args:
        BaseModel (Class): The base model
    """
    userId: str
    setting: dict

# Event Model
class Event(BaseModel):
    """
    Event model used in this API

    Args:
        BaseModel (Class): The base model
    """
    date: str
    events: dict


# ---------- USER FUNCTION ----------

@app.get('/api/setting/{id_user}/', response_model=User)
def get_user_settings(id_user: str):
    """
    A function to bring the user's setting through the generated key to Virgilio.    

     Raises:
        HTTPException: Error
    Returns:
        JsonResponse: the result of request
    """
    result = users_collection.find_one({"userId": str(id_user)}, {"_id": 0, "userId": 0})
    if result is None:
        raise HTTPException(status_code=404, detail="User not found")
    user_dict = dict(result)
    return JSONResponse(content=user_dict, status_code=200)

def check_email_pass(list_of_events):
    """
    A function that checks whether a list contains an email and password or not

    Args:
        list (list): An list of data

    Returns:
        bool: False or None
    """
    for i in list_of_events:
        if i == "":
            return False

@app.post('/api/setting/modify/{id_user}/', response_model=User)
def new_setting(id_user: str, new_setting: dict):
    """
    This function updates all the setting of Virgil specifying the key of the user and
    sends a payload in json and skips the empty values.
    
    Returns:
        dict: Json format file       
    """
    updates = {f"setting.{key}": value for key, value in new_setting.items() if value != ""}
    query = {"userId": str(id_user)}
    value = {"$set": updates}
    users_collection.update_many(query, value)
    return get_user_settings(id_user)


# ---- CALENDAR ----
@app.put('/api/createUser', response_model=User, status_code=201)
def create_user():
    """
    This function creates a new user which is entered into the database by the 
    simple random key generated randomly and the setting base    

    Returns:
        dict: The dict with the user id and the settings
    """

    key = secrets.token_hex(16)
    users_collection.insert_one({
        "userId": key,
        "setting": setting
    })

    return {"userId": key, "setting": setting}

# ---------- CALENDAR FUNCTION ----------

@app.get('/api/calendar/{id_user}/', response_model=dict)
def get_events(id_user: str):
    """
    Get all the events from a user by the id

    Args:
        id (str): The id of user

    Raises:
        HTTPException: _description_

    Returns:
        list: List of events from the user
    """
    result = calendar_collection.find_one({"userId": str(id_user)},{"_id":0,"userId":0})
    if result is None:
        raise HTTPException(status_code=404, detail="User not found")
    return result

@app.put('/api/calendar/createUser/{id_user}/', status_code=201)
def create_user_calendar(id_user: str):
    """
    Create the profile in the db for manage the events of a user

    Args:
        id (str): id of user

    Returns:
        id_user: the id of user
    """
    calendar_collection.insert_one({"userId": id_user}) # Prepare the user for give event
    return id_user

@app.put('/api/calendar/createEvent/{id_user}/{date}/', status_code=201)
def create_event(id_user: str, date: str, events: List[str]):
    # Cambiato events: Event a events: List[str]
    """
    Create an event to save it into the database

    Args:
        id (str): The id of user
        date (str): the date of events
        events (List[str]): the description of events

    Returns:
        dict: The final result of modify
    """

    result = calendar_collection.find_one({"userId": id_user}, {date: 1})
    query = {"userId": id_user}
    if result is None or date not in result:
        # Utilizziamo direttamente il payload JSON per l'aggiornamento
        value = {"$set": {date: events}}
        result = calendar_collection.update_one(query, value)
    else:
        # Utilizziamo direttamente il payload JSON per l'aggiornamento
        value = {"$addToSet": {date: {"$each": events}}}
        result = calendar_collection.update_many(query, value)  # Aggiungi evento
    return value


def get_formatted_date():
    """
    Get today's format date

    Returns:
        str: The date formatted
    """
    today = datetime.datetime.today()
    yesterday = today.date() + datetime.timedelta(days=-1)
    yesterday = yesterday.strftime("%d-%m-%Y")
    yesterday = yesterday.split("-")
    yesterday[1] = int(yesterday[1])
    yesterday[1] = str(yesterday[1])
    yesterday[0] = int(yesterday[0])
    yesterday[0] = str(yesterday[0])
    yesterday = "-".join(yesterday)

    return yesterday

@app.put('/api/calendar/deleteEvent/{id_user}/', status_code=201)
def delete_event(id_user: str):
    """
    Delete all old events from the day and update the collection

    Args:
        id (str): id of user

    Returns:
        dict:Element deleted
    """
    yesterday = get_formatted_date()
    result = calendar_collection.find_one({"userId": id_user})
    query = {"userId": id_user}
    if result is None or yesterday not in result:
        return {"Delete": "No events yesterday"}, 202
    value = {"$unset": {yesterday: 1}}
    result = calendar_collection.update_one(query, value) # Add event
    return value


1			"""
2			Main file for the API
3			"""
4			import datetime
5			import json
6			import secrets
7			import os
8			from typing import List
9			import functools
10
11			from fastapi import FastAPI, HTTPException
12			from pydantic import BaseModel
13			from pymongo import MongoClient
14			from fastapi.responses import JSONResponse
15
16			# Take the url for connection to MongoDB
17			urlMongo = os.getenv('MONGO_URL')
18
19			# Init the client of Mongo
20			client = MongoClient(urlMongo)
21
22			db = client.virgilUsers
23			users_collection = db.users
24			users_collection.create_index("userId", unique=True)
25			calendar_collection = db.calendarEvent
26
27			app = FastAPI()
28
29			# Take the base of setting
30			@functools.lru_cache(maxsize=None)
31			def get_cached_setting():
32			"""
33			Get all settings from database and cache it in memory
34
35			Returns:
36			json: The default settings
37			"""
38			with open('setting.json', 'r',encoding='utf-8') as file:
39			return json.load(file)
40
41			# Modifica la chiamata nel codice originale
42			setting = get_cached_setting()
43
44			# User Model
45			class User(BaseModel):
46			"""
47			User model used in this API
48
49			Args:
50			BaseModel (Class): The base model
51			"""
52			userId: str
53			setting: dict
54
55			# Event Model
56			class Event(BaseModel):
57			"""
58			Event model used in this API
59
60			Args:
61			BaseModel (Class): The base model
62			"""
63			date: str
64			events: dict
65
66
67			# ---------- USER FUNCTION ----------
68
69			@app.get('/api/setting/{id_user}/', response_model=User)
70			def get_user_settings(id_user: str):
71			"""
72			A function to bring the user's setting through the generated key to Virgilio.
73
74			Raises:
75			HTTPException: Error
76			Returns:
77			JsonResponse: the result of request
78			"""
79			result = users_collection.find_one({"userId": str(id_user)}, {"_id": 0, "userId": 0})
80			if result is None:
81			raise HTTPException(status_code=404, detail="User not found")
82			user_dict = dict(result)
83			return JSONResponse(content=user_dict, status_code=200)
84
85			def check_email_pass(list_of_events):
86			"""
87			A function that checks whether a list contains an email and password or not
88
89			Args:
90			list (list): An list of data
91
92			Returns:
93			bool: False or None
94			"""
95			for i in list_of_events:
96			if i == "":
97			return False
98
99			@app.post('/api/setting/modify/{id_user}/', response_model=User)
100			def new_setting(id_user: str, new_setting: dict):
101			"""
102			This function updates all the setting of Virgil specifying the key of the user and
103			sends a payload in json and skips the empty values.
104
105			Returns:
106			dict: Json format file
107			"""
108			updates = {f"setting.{key}": value for key, value in new_setting.items() if value != ""}
109			query = {"userId": str(id_user)}
110			value = {"$set": updates}
111			users_collection.update_many(query, value)
112			return get_user_settings(id_user)
113
114
115			# ---- CALENDAR ----
116			@app.put('/api/createUser', response_model=User, status_code=201)
117			def create_user():
118			"""
119			This function creates a new user which is entered into the database by the
120			simple random key generated randomly and the setting base
121
122			Returns:
123			dict: The dict with the user id and the settings
124			"""
125
126			key = secrets.token_hex(16)
127			users_collection.insert_one({
128			"userId": key,
129			"setting": setting
130			})
131
132			return {"userId": key, "setting": setting}
133
134			# ---------- CALENDAR FUNCTION ----------
135
136			@app.get('/api/calendar/{id_user}/', response_model=dict)
137			def get_events(id_user: str):
138			"""
139			Get all the events from a user by the id
140
141			Args:
142			id (str): The id of user
143
144			Raises:
145			HTTPException: _description_
146
147			Returns:
148			list: List of events from the user
149			"""
150			result = calendar_collection.find_one({"userId": str(id_user)},{"_id":0,"userId":0})
151			if result is None:
152			raise HTTPException(status_code=404, detail="User not found")
153			return result
154
155			@app.put('/api/calendar/createUser/{id_user}/', status_code=201)
156			def create_user_calendar(id_user: str):
157			"""
158			Create the profile in the db for manage the events of a user
159
160			Args:
161			id (str): id of user
162
163			Returns:
164			id_user: the id of user
165			"""
166			calendar_collection.insert_one({"userId": id_user}) # Prepare the user for give event
167			return id_user
168
169			@app.put('/api/calendar/createEvent/{id_user}/{date}/', status_code=201)
170			def create_event(id_user: str, date: str, events: List[str]):
171			# Cambiato events: Event a events: List[str]
172			"""
173			Create an event to save it into the database
174
175			Args:
176			id (str): The id of user
177			date (str): the date of events
178			events (List[str]): the description of events
179
180			Returns:
181			dict: The final result of modify
182			"""
183
184			result = calendar_collection.find_one({"userId": id_user}, {date: 1})
185			query = {"userId": id_user}
186			if result is None or date not in result:
187			# Utilizziamo direttamente il payload JSON per l'aggiornamento
188			value = {"$set": {date: events}}
189			result = calendar_collection.update_one(query, value)
190			else:
191			# Utilizziamo direttamente il payload JSON per l'aggiornamento
192			value = {"$addToSet": {date: {"$each": events}}}
193			result = calendar_collection.update_many(query, value) # Aggiungi evento
194			return value
195
196
197			def get_formatted_date():
198			"""
199			Get today's format date
200
201			Returns:
202			str: The date formatted
203			"""
204			today = datetime.datetime.today()
205			yesterday = today.date() + datetime.timedelta(days=-1)
206			yesterday = yesterday.strftime("%d-%m-%Y")
207			yesterday = yesterday.split("-")
208			yesterday[1] = int(yesterday[1])
209			yesterday[1] = str(yesterday[1])
210			yesterday[0] = int(yesterday[0])
211			yesterday[0] = str(yesterday[0])
212			yesterday = "-".join(yesterday)
213
214			return yesterday
215
216			@app.put('/api/calendar/deleteEvent/{id_user}/', status_code=201)
217			def delete_event(id_user: str):
218			"""
219			Delete all old events from the day and update the collection
220
221			Args:
222			id (str): id of user
223
224			Returns:
225			dict:Element deleted
226			"""
227			yesterday = get_formatted_date()
228			result = calendar_collection.find_one({"userId": id_user})
229			query = {"userId": id_user}
230			if result is None or yesterday not in result:
231			return {"Delete": "No events yesterday"}, 202
232			value = {"$unset": {yesterday: 1}}
233			result = calendar_collection.update_one(query, value) # Add event
234			return value
235

Retr0100 / VirgilAPI

main.create_event() A last analyzed 2023-09-04 23:03 UTC

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like

main.create_event() A
last analyzed 2023-09-04 23:03 UTC