-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathservices.py
More file actions
45 lines (37 loc) · 1.41 KB
/
services.py
File metadata and controls
45 lines (37 loc) · 1.41 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
from typing import TYPE_CHECKING
import database as database
import schemas as schemas
import models as models
if TYPE_CHECKING:
from sqlalchemy.orm import Session
def add_tables():
return database.Base.metadata.create_all(bind=database.engine)
def get_db():
db = database.SessionLocal()
try:
yield db
finally:
db.close()
async def create_contact(contact: schemas.CreateContact, db: "Session") -> schemas.Contact:
contact = models.Contact(**contact.dict())
db.add(contact)
db.commit()
db.refresh(contact)
return schemas.Contact.from_orm(contact)
async def get_all_contacts(db: "Session") -> list[schemas.Contact]:
contacts = db.query(models.Contact).all()
return list(map(schemas.Contact.from_orm, contacts))
async def get_contact(contact_id: int, db: "Session"):
contact = db.query(models.Contact).filter(models.Contact.id == contact_id).first()
return contact
async def delete_contact(contact: models.Contact, db: "Session"):
db.delete(contact)
db.commit()
async def update_contact(contact: models.Contact, contact_data: schemas.CreateContact, db: "Session") -> schemas.Contact:
contact.first_name = contact_data.first_name
contact.last_name = contact_data.last_name
contact.email = contact_data.email
contact.phone_number = contact_data.phone_number
db.commit()
db.refresh(contact)
return schemas.Contact.from_orm(contact)