Coverage for src/routes/contacts.py: 51%
53 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-01 22:29 +0200
1from fastapi import APIRouter, HTTPException, Depends, status, Query
3from sqlalchemy.ext.asyncio import AsyncSession
5from src.entity.models import User
6from src.database.db import get_db
7from src.repository import contacts as repositories_contacts
8from src.schemas.contact import ContactSchema, ContactUpdateSchema, ContactResponseSchema
9from src.services.auth import auth_serviсe
11from fastapi_limiter.depends import RateLimiter
13router = APIRouter(prefix="/contacts", tags=["contacts"])
16@router.get(
17 "/",
18 response_model=list[ContactResponseSchema],
19 description="No more than 2 requests per minute",
20 dependencies=[Depends(RateLimiter(times=2, seconds=60))],
21)
22async def get_contacts(
23 limit: int = Query(default=10, ge=1, le=500),
24 offset: int = Query(default=0, ge=0),
25 db: AsyncSession = Depends(get_db),
26 user: User = Depends(auth_serviсe.get_current_user)
27):
28 """
29 Get a list of contacts.
31 Parameters:
32 - limit (int): The maximum number of contacts to retrieve. Default is 10, minimum is 1, and maximum is 500.
33 - offset (int): The index of the first contact to retrieve. Default is 0.
34 - db (AsyncSession): The database session to use for the operation.
35 - user (User): The authenticated user making the request.
37 Returns:
38 - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts.
39 """
40 contacts = await repositories_contacts.get_contacts(limit, offset, db, user)
41 return contacts
44@router.get("/{contact_id}", response_model=ContactResponseSchema)
45async def get_contact(
46 contact_id: int,
47 db: AsyncSession = Depends(get_db),
48 user: User = Depends(auth_serviсe.get_current_user),
49):
50 """
51 Get a specific contact by its ID.
53 Parameters:
54 - contact_id (int): The ID of the contact to retrieve.
55 - db (AsyncSession): The database session to use for the operation.
56 - user (User): The authenticated user making the request.
58 Returns:
59 - ContactResponseSchema: The data of the specific contact.
60 - HTTPException: If the contact with the specified ID is not found.
61 """
62 contact = await repositories_contacts.get_contact(contact_id, db, user)
63 if contact is None:
64 raise HTTPException(
65 status_code=status.HTTP_404_NOT_FOUND, detail="Contact not found"
66 )
67 return contact
70@router.post(
71 "/",
72 response_model=ContactResponseSchema,
73 status_code=status.HTTP_201_CREATED,
74 description="No more than 2 contacts per one minute",
75 dependencies=[Depends(RateLimiter(times=2, seconds=60))],
76)
77async def create_contact(
78 body: ContactSchema,
79 db: AsyncSession = Depends(get_db),
80 user: User = Depends(auth_serviсe.get_current_user),
81):
82 """
83 Create a new contact.
85 Parameters:
86 - body (ContactSchema): The data of the new contact to be created.
87 - db (AsyncSession): The database session to use for the operation.
88 - user (User): The authenticated user making the request.
90 Returns:
91 - ContactResponseSchema: The newly created contact data.
92 """
93 contact = await repositories_contacts.create_contact(body, db, user)
94 return contact
97@router.delete("/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
98async def delete_contact(
99 contact_id: int,
100 db: AsyncSession = Depends(get_db),
101 user: User = Depends(auth_serviсe.get_current_user),
102):
103 """
104 Delete a contact by its ID.
106 Parameters:
107 - contact_id (int): The ID of the contact to delete.
108 - db (AsyncSession): The database session to use for the operation.
109 - user (User): The authenticated user making the request.
111 Returns:
112 - ContactResponseSchema: The deleted contact data.
113 - HTTPException: If the contact with the specified ID is not found.
114 """
115 contact = await repositories_contacts.delete_contact(contact_id, db, user)
116 if contact is None:
117 raise HTTPException(
118 status_code=status.HTTP_404_NOT_FOUND, detail="Contact is not found"
119 )
120 return contact
123@router.put("/{contact_id}", response_model=ContactResponseSchema)
124async def update_contact(
125 body: ContactUpdateSchema,
126 contact_id: int,
127 db: AsyncSession = Depends(get_db),
128 user: User = Depends(auth_serviсe.get_current_user),
129):
130 """
131 Update an existing contact.
133 Parameters:
134 - body (ContactUpdateSchema): The updated contact data.
135 - contact_id (int): The ID of the contact to update.
136 - db (AsyncSession): The database session to use for the operation.
137 - user (User): The authenticated user making the request.
139 Returns:
140 - ContactResponseSchema: The updated contact data.
141 - HTTPException: If the contact with the specified ID is not found.
142 """
143 contact = await repositories_contacts.update_contact(contact_id, body, db, user)
144 if contact is None:
145 raise HTTPException(
146 status_code=status.HTTP_404_NOT_FOUND, detail="Contact is not found"
147 )
148 return contact
151@router.get("/email/{contact_email}", response_model=ContactResponseSchema)
152async def get_contact_by_email(
153 contact_email: str,
154 db: AsyncSession = Depends(get_db),
155 user: User = Depends(auth_serviсe.get_current_user),
156):
157 """
158 Get a contact by their email address.
160 Parameters:
161 - contact_email (str): The email address of the contact to search for.
162 - db (AsyncSession): The database session to use for the operation.
163 - user (User): The authenticated user making the request.
165 Returns:
166 - ContactResponseSchema: A ContactResponseSchema object representing the contact with the specified email address.
167 - HTTPException: If the contact with the specified email address is not found.
168 """
169 contact = await repositories_contacts.get_contact_by_email(contact_email, db, user)
170 if contact is None:
171 raise HTTPException(
172 status_code=status.HTTP_404_NOT_FOUND, detail="Contact not found"
173 )
174 return contact
177@router.get("/last_name/{contact_last_name}", response_model=list[ContactResponseSchema])
178async def get_contact_by_last_name(
179 contact_last_name: str,
180 db: AsyncSession = Depends(get_db),
181 user: User = Depends(auth_serviсe.get_current_user),
182):
183 """
184 Get all contacts with the specified last name.
186 Parameters:
187 - contact_last_name (str): The last name to search for in the contacts.
188 - db (AsyncSession): The database session to use for the operation.
189 - user (User): The authenticated user making the request.
191 Returns:
192 - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with the specified last name.
193 """
194 contacts = await repositories_contacts.get_contact_by_last_name(
195 contact_last_name, db, user
196 )
197 return contacts
200@router.get("/birthday/{contact_birthday}", response_model=list[ContactResponseSchema])
201async def get_contact_by_birthday(
202 contact_birthday: str,
203 db: AsyncSession = Depends(get_db),
204 user: User = Depends(auth_serviсe.get_current_user),
205):
206 """
207 Get all contacts with birthdays on the specified date.
209 Parameters:
210 - contact_birthday (str): The date in the format 'YYYY-MM-DD' to search for contacts with birthdays.
211 - db (AsyncSession): The database session to use for the operation.
212 - user (User): The authenticated user making the request.
214 Returns:
215 - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with birthdays on the specified date.
216 """
217 contacts = await repositories_contacts.get_contact_by_birthday(
218 contact_birthday, db, user
219 )
220 return contacts
223@router.get("/birthdays/", response_model=list[ContactResponseSchema])
224async def get_birthdays(
225 db: AsyncSession = Depends(get_db),
226 user: User = Depends(auth_serviсe.get_current_user),
227):
228 """
229 Get all contacts with birthdays within 7 days.
231 Parameters:
232 - db (AsyncSession): The database session to use for the operation.
233 - user (User): The authenticated user making the request.
235 Returns:
236 - list[ContactResponseSchema]: A list of ContactResponseSchema objects representing the contacts with birthdays within 7 days.
237 """
238 contacts = await repositories_contacts.get_birthdays(db, user)
239 return contacts