Coverage for toardb/contacts/contacts.py: 95%
82 statements
« prev ^ index » next coverage.py v7.11.0, created at 2025-11-03 20:32 +0000
1# SPDX-FileCopyrightText: 2021 Forschungszentrum Jülich GmbH
2# SPDX-License-Identifier: MIT
4"""
5Simple API for contacts management
6"""
8from typing import List, Union, Literal
9from fastapi import APIRouter, Depends, HTTPException, Body, Request
11from sqlalchemy.orm import Session
12from . import crud, schemas
13from toardb.utils.database import ToarDbSession, get_db
15router = APIRouter()
17# plain views to post and get contacts
19# 1. persons
21#get all entries of table persons
22@router.get('/contacts/persons/', response_model=List[schemas.Person], response_model_exclude_none=True, response_model_exclude_unset=True)
23def get_all_persons(request: Request, db: Session = Depends(get_db)):
24 persons = crud.get_all_persons(db, path_params=request.path_params, query_params=request.query_params)
25 return persons
27#get all metadata of one person
28@router.get('/contacts/persons/id/{person_id}', response_model=schemas.Person)
29def get_person(person_id: int, db: Session = Depends(get_db)):
30 db_person = crud.get_person(db, person_id=person_id)
31 if db_person is None:
32 raise HTTPException(status_code=404, detail="Person not found.")
33 return db_person
35@router.get('/contacts/persons/{name}', response_model=schemas.Person)
36def get_person(name: str, db: Session = Depends(get_db)):
37 db_person = crud.get_person_by_name(db, name=name)
38 if db_person is None:
39 raise HTTPException(status_code=404, detail="Person not found.")
40 return db_person
42@router.post('/contacts/persons/', response_model=schemas.Person)
43def create_person(person: schemas.PersonCreate = Body(..., embed = True), db: Session = Depends(get_db)):
44 db_person = crud.get_person_by_name(db, name=person.name)
45 if db_person:
46 raise HTTPException(status_code=400, detail="Person already registered.")
47 return crud.create_person(db=db, person=person)
50# 2. organisations
52#get all entries of table organisations
53@router.get('/contacts/organisations/', response_model=List[schemas.Organisation])
54def get_all_organisations(offset: int = 0, limit: int | Literal["None"] = 10, db: Session = Depends(get_db)):
55 if limit == "None":
56 limit = None
57 organisations = crud.get_all_organisations(db, offset=offset, limit=limit)
58 return organisations
60#get all metadata of one organisation
61@router.get('/contacts/organisations/id/{organisation_id}', response_model=schemas.Organisation)
62def get_organisation(organisation_id: int, db: Session = Depends(get_db)):
63 db_organisation = crud.get_organisation (db, organisation_id=organisation_id)
64 if db_organisation is None:
65 raise HTTPException(status_code=404, detail="Organisation not found.")
66 return db_organisation
68@router.get('/contacts/organisations/{name}', response_model=schemas.Organisation)
69def get_organisation(name: str, db: Session = Depends(get_db)):
70 db_organisation = crud.get_organisation_by_name(db, name=name)
71 if db_organisation is None:
72 raise HTTPException(status_code=404, detail="Organisation not found.")
73 return db_organisation
75@router.post('/contacts/organisations/', response_model=schemas.Organisation)
76def create_organisation(organisation: schemas.OrganisationCreate = Body(..., embed = True), db: Session = Depends(get_db)):
77 db_organisation = crud.get_organisation_by_longname(db, longname=organisation.longname)
78 if db_organisation:
79 raise HTTPException(status_code=400, detail="Organisation already registered.")
80 return crud.create_organisation(db=db, organisation=organisation)
82# 3. combined (person + organisation)
83#get all entries of table contacts
84@router.get('/contacts/', response_model=List[schemas.Contact])
85def get_all_contacts(offset: int = 0, limit: int | Literal["None"] = 10, db: Session = Depends(get_db)):
86 if limit == "None":
87 limit = None
88 contacts = crud.get_all_contacts(db, offset=offset, limit=limit)
89 return contacts
91#get a special entry of table contacts (given by ID)
92@router.get('/contacts/id/{contact_id}', response_model=Union[schemas.Organisation,schemas.Person])
93def get_all_contacts(contact_id: int, db: Session = Depends(get_db)):
94 contact = crud.get_contact(db, contact_id=contact_id)
95 return contact
97#get a special entry of table contacts (given by its organisation (long) name)
98@router.get('/contacts/orga_name/{name}', response_model=schemas.Contact)
99def get_all_contacts(name: str, db: Session = Depends(get_db)):
100 contact = crud.get_contact_by_orga_name(db, name=name)
101 if contact is None:
102 raise HTTPException(status_code=404, detail="Contact not found.")
103 return contact
105#get a special entry of table contacts (given by its person name)
106@router.get('/contacts/person_name/{name}', response_model=schemas.Contact)
107def get_all_contacts(name: str, db: Session = Depends(get_db)):
108 contact = crud.get_contact_by_person_name(db, name=name)
109 return contact
112#get a special entry of table contacts (given by its person email)
113@router.get('/contacts/person_email/{email}', response_model=schemas.Contact)
114def get_all_contacts(email: str, db: Session=Depends(get_db)):
115 contact = crud.get_contact_by_person_email(db, email=email)
116 return contact
119#create new contact (either only person/organisation or both)
120@router.post('/contacts/')
121def create_contact(person_name: str = None, organisation_longname: str = None, db: Session = Depends(get_db)):
122 if not person_name and not organisation_longname:
123 raise HTTPException(status_code=400, detail="No contact data given.")
124 contact = crud.create_contact(db,person_name=person_name,organisation_longname=organisation_longname)
125 return contact