Coverage for src/repository/contacts.py: 88%
75 statements
« prev ^ index » next coverage.py v7.5.4, created at 2024-07-01 22:29 +0200
1from fastapi import HTTPException, status
2from sqlalchemy import Date, select
3from sqlalchemy.ext.asyncio import AsyncSession
5from sqlalchemy import and_
7from src.entity.models import Contact, User
8from src.schemas.contact import ContactSchema, ContactUpdateSchema
11async def get_contacts(limit: int, offset: int, db: AsyncSession, user: User):
12 """
13 Retrieves a list of contacts for the specified user within the specified limit and offset.
15 Args:
16 - limit (int): The maximum number of contacts to retrieve.
17 - offset (int): The index from which to start retrieving contacts.
18 - db (AsyncSession): The database session.
19 - user (User): The user for whom the contacts are being retrieved.
21 Returns:
22 - list: A list of Contact objects for the specified user within the specified limit and offset.
23 """
24 stmt = select(Contact).filter_by(user=user).offset(offset).limit(limit)
25 result = await db.execute(stmt)
26 contacts = result.scalars().all()
27 return contacts
30async def get_contact(contact_id: int, db: AsyncSession, user: User):
31 """
32 Retrieves a specific contact by its id for the specified user.
34 Args:
35 - contact_id (int): The id of the contact to retrieve.
36 - db (AsyncSession): The database session.
37 - user (User): The user for whom the contact is being retrieved.
39 Returns:
40 - Contact or None: The specific contact object for the specified user and contact_id, or None if not found.
41 """
42 stmt = select(Contact).filter_by(id=contact_id, user=user)
43 result = await db.execute(stmt)
44 contact = result.scalar_one_or_none()
45 return contact
49async def create_contact(body: ContactSchema, db: AsyncSession, user: User):
50 """
51 Creates a new contact for the specified user.
53 Args:
54 - body (ContactSchema): The schema containing the details of the new contact.
55 - db (AsyncSession): The database session.
56 - user (User): The user for whom the contact is being created.
58 Raises:
59 - HTTPException: If a contact with the same first name, last name, and email already exists.
61 Returns:
62 - Contact: The newly created contact object.
63 """
64 # Перевірка наявності контакту в базі по імені, призвіщу
65 existing_contact = await db.execute(
66 select(Contact).filter_by(
67 first_name=body.first_name,
68 last_name=body.last_name,
69 email=body.email
70 )
71 )
72 existing_contact = existing_contact.fetchone()
73 if existing_contact:
74 raise HTTPException(
75 status_code=400,
76 detail="Contact is already exist",
77 )
78 contact = Contact(**body.model_dump(), user=user)
79 db.add(contact)
80 await db.commit()
81 await db.refresh(contact)
82 return contact
84async def delete_contact(contact_id: int, db: AsyncSession, user: User):
85 """
86 Deletes a specific contact by its id for the specified user.
88 Args:
89 - contact_id (int): The id of the contact to delete.
90 - db (AsyncSession): The database session.
91 - user (User): The user for whom the contact is being deleted.
93 Returns:
94 - Contact or None: The specific contact object for the specified user and contact_id, or None if not found.
95 """
96 stmt = select(Contact).filter_by(id=contact_id, user=user)
97 result = await db.execute(stmt)
98 contact = result.scalar_one_or_none()
99 if contact:
100 await db.delete(contact)
101 await db.commit()
102 return contact
105async def update_contact(
106 contact_id: int, # The id of the contact to update.
107 body: ContactUpdateSchema, # The schema containing the details of the updated contact.
108 db: AsyncSession, # The database session.
109 user: User, # The user for whom the contact is being updated.
110) -> Contact: # Returns the updated contact object.
111 """
112 Updates a specific contact by its id for the specified user.
114 Args:
115 - contact_id (int): The id of the contact to update.
116 - body (ContactUpdateSchema): The schema containing the details of the updated contact.
117 - db (AsyncSession): The database session.
118 - user (User): The user for whom the contact is being updated.
120 Returns:
121 - Contact: The updated contact object.
122 """
123 stmt = select(Contact).filter_by(id=contact_id, user=user)
124 result = await db.execute(stmt)
125 contact = result.scalar_one_or_none()
126 if contact is None:
127 return None
128 for key, value in body.model_dump().items():
129 setattr(contact, key, value)
130 await db.commit()
131 await db.refresh(contact)
132 return contact
135async def get_contact_by_email(contact_email: str, db: AsyncSession, user: User):
136 """
137 Retrieves a specific contact by its email for the specified user.
139 Args:
140 - contact_email (str): The email of the contact to retrieve.
141 - db (AsyncSession): The database session.
142 - user (User): The user for whom the contact is being retrieved.
144 Returns:
145 - Contact or None: The specific contact object for the specified user and contact_email, or None if not found.
146 """
147 stmt = select(Contact).filter_by(email=contact_email, user=user)
148 result = await db.execute(stmt)
149 contact = result.scalar_one_or_none()
150 if not contact:
151 raise HTTPException(
152 status_code=status.HTTP_404_NOT_FOUND,
153 detail="Contact is not found",
154 )
155 return contact
158async def get_contact_by_last_name(
159 contact_last_name: str, # The last name of the contact to retrieve.
160 db: AsyncSession, # The database session.
161 user: User, # The user for whom the contact is being retrieved.
162) -> list: # Returns a list of Contact objects for the specified user and last name.
163 """
164 Retrieves a list of contacts for the specified user with the given last name.
166 Args:
167 - contact_last_name (str): The last name of the contact to retrieve.
168 - db (AsyncSession): The database session.
169 - user (User): The user for whom the contact is being retrieved.
171 Returns:
172 - list: A list of Contact objects for the specified user and last name.
173 """
174 stmt = select(Contact).filter_by(last_name=contact_last_name, user=user)
175 result = await db.execute(stmt)
176 contacts = result.scalars().all()
177 if not contacts:
178 raise HTTPException(
179 status_code=status.HTTP_404_NOT_FOUND,
180 detail="Contacts not found",
181 )
182 return contacts
185async def get_contact_by_birthday(contact_birthday: Date, db: AsyncSession, user: User):
186 """
187 Retrieves a list of contacts for the specified user with the given birthday.
189 Args:
190 - contact_birthday (Date): The birthday of the contact to retrieve.
191 - db (AsyncSession): The database session.
192 - user (User): The user for whom the contact is being retrieved.
194 Returns:
195 - list: A list of Contact objects for the specified user and birthday.
197 Raises:
198 - HTTPException: If no contacts are found with the given birthday and user.
199 """
200 stmt = select(Contact).filter_by(birthday=contact_birthday, user=user)
201 result = await db.execute(stmt)
202 contacts = result.scalars().all()
203 if not contacts:
204 raise HTTPException(
205 status_code=status.HTTP_404_NOT_FOUND,
206 detail="Contacts not found",
207 )
208 return contacts
211async def get_birthdays(db: AsyncSession, user: User):
212 """
213 Retrieves a list of contacts for the specified user with birthdays within the next 7 days.
215 Args:
216 - db (AsyncSession): The database session.
217 - user (User): The user for whom the contacts are being retrieved.
219 Returns:
220 - list: A list of Contact objects for the specified user with birthdays within the next 7 days.
222 Raises:
223 - HTTPException: If no contacts are found with the given birthday and user.
224 """
225 from datetime import datetime, timedelta
226 from sqlalchemy import select, and_
227 current_date = datetime.now()
228 end_date = current_date + timedelta(days=7)
229 stmt = select(Contact).filter(
230 and_(
231 Contact.birthday.between(current_date.date(), end_date.date()),
232 Contact.user_id
233 == user.id, # Фільтруємо за ідентифікатором поточного користувача
234 )
235 )
237 result = await db.execute(stmt)
238 contacts = result.scalars().all()
239 return contacts