sr2/cloud-api
6
0
Fork 1
Code Issues 8 Pull requests 2 Projects Releases Packages Wiki Activity Actions

feat: contact model restructure

Browse source 
Blank contacts are now generated on org creation and assigned to each contact type. These contacts are linked to the org, only accessible to the org, and removed when the org is removed.

With this all contact endpoints have been removed. Contact manipulation is done via the org only.
This commit is contained in:
Chris Milne 2026-05-25 15:15:50 +01:00
parent 707482adc2
commit 2b6d923ae1
7 changed files with 146 additions and 165 deletions
Download patch file Download diff file Expand all files Collapse all files

34
.alembic/versions/2026-05-25_contact_model_changes.py Normal file
View file

@ -0,0 +1,34 @@
"""Contact model changes
Revision ID: 8132c4b88665
Revises: a147965e644e
Create Date: 2026-05-25 13:09:22.635058
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = '8132c4b88665'
down_revision: Union[str, Sequence[str], None] = 'a147965e644e'
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
"""Upgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('contact', sa.Column('org_id', sa.Integer(), nullable=False))
op.create_foreign_key(None, 'contact', 'organisation', ['org_id'], ['id'], ondelete='CASCADE')
# ### end Alembic commands ###
def downgrade() -> None:
"""Downgrade schema."""
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint(None, 'contact', type_='foreignkey')
op.drop_column('contact', 'org_id')
# ### end Alembic commands ###

4
src/contact/models.py
View file

@ -5,7 +5,7 @@ Models:
- Contact: id[pk], email, first_name, last_name, phonenumber, vat_number
street_address, post_office_box_number, address_locality, country_code, address_region, postal_code
"""
from sqlalchemy import Column, Integer, String
from sqlalchemy import Column, Integer, String, ForeignKey
from src.database import Base
@ -27,3 +27,5 @@ class Contact(Base):
country_code = Column(String) # Eg GB
address_region = Column(String, default=None, nullable=True)
postal_code = Column(String)
org_id = Column(Integer, ForeignKey("organisation.id", ondelete="CASCADE"), nullable=False)

102
src/contact/router.py
View file

@ -9,110 +9,10 @@ Endpoints:
- [patch]/{contact_id} - Updates the details of an existing contact
- [delete]/{contact_id} - Deletes a contact by ID
"""
from typing import Annotated
from fastapi import APIRouter
from fastapi import APIRouter, HTTPException
from fastapi.params import Path
from sqlalchemy import or_
from src.contact.schemas import ContactContactGetResponse, ContactAddressGetResponse, ContactContactPostRequest, \
ContactUpdateRequest, ContactOrgGetResponse
from src.contact.models import Contact
from src.database import db_dependency
from src.organisation.models import Organisation as Org
from src.organisation.constants import ContactType
router = APIRouter(
prefix="/contact",
tags=["contact"],
)
@router.get("/{contact_id}", response_model=ContactContactGetResponse)
async def get_contact_details_by_id(contact_id: int, db: db_dependency):
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
return contact_model
@router.get("/{contact_id}/address", response_model=ContactAddressGetResponse)
async def get_contact_address_by_id(contact_id: int, db: db_dependency):
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
return contact_model
@router.post("/")
async def create_contact(db: db_dependency, contact_request: ContactContactPostRequest):
contact_model = Contact(**contact_request.model_dump())
db.add(contact_model)
db.commit()
@router.patch("/{contact_id}")
async def update_contact(db: db_dependency, contact_request: ContactUpdateRequest, contact_id: Annotated[int, Path(gt=0)]):
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
update_data = contact_request.model_dump(exclude_none=True)
for key, value in update_data.items():
if hasattr(contact_model, key):
setattr(contact_model, key, value)
else:
raise HTTPException(status_code=422, detail="Invalid keys in update request")
db.add(contact_model)
db.commit()
@router.delete("/{contact_id}")
async def delete_contact(db: db_dependency, contact_id: Annotated[int, Path(gt=0)]):
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
db.delete(contact_model)
db.commit()
@router.get("/{contact_id}/orgs", response_model=list[ContactOrgGetResponse])
async def get_contact_orgs(db: db_dependency, contact_id: Annotated[int, Path(gt=0)]):
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
org_models = (db.query(Org).filter(
or_(
Org.owner_contact_id == contact_id,
Org.billing_contact_id == contact_id,
Org.security_contact_id == contact_id
)
).all())
response = []
for org in org_models:
types=[]
if org.owner_contact_id == contact_id:
types.append(ContactType.OWNER)
if org.billing_contact_id == contact_id:
types.append(ContactType.BILLING)
if org.security_contact_id == contact_id:
types.append(ContactType.SECURITY)
org_response_model = ContactOrgGetResponse(
name=str(org.name),
contact_types=types,
)
response.append(org_response_model)
return response

14
src/contact/schemas.py
View file

@ -7,12 +7,24 @@ Models:
"""
from typing import Optional
from pydantic import EmailStr
from pydantic import EmailStr, ConfigDict
from src.organisation.constants import ContactType
from src.schemas import CustomBaseModel
class ContactAddress(CustomBaseModel):
model_config = ConfigDict(from_attributes=True, extra="ignore")
post_office_box_number: Optional[str] = None
street_address: Optional[str] = None
street_address_line_2: Optional[str] = None
locality: Optional[str] = None
address_region: Optional[str] = None
country_code: Optional[str] = None
postal_code: Optional[str] = None
class ContactContactGetResponse(CustomBaseModel):
email: str
first_name: str

4
src/organisation/models.py
View file

@ -39,6 +39,10 @@ class Organisation(Base):
def root_user_email(self):
return self.root_user_rel.email if self.root_user_rel else None
billing_contact_rel = relationship("Contact", foreign_keys=[billing_contact_id])
security_contact_rel = relationship("Contact", foreign_keys=[security_contact_id])
owner_contact_rel = relationship("Contact", foreign_keys=[owner_contact_id])
class OrgUsers(Base):
__tablename__ = "orgusers"

111
src/organisation/router.py
View file

@ -19,6 +19,7 @@ from fastapi.params import Path, Query
from sqlalchemy.sql import exists
from src.contact.schemas import ContactAddress
from src.database import db_dependency
from src.contact.models import Contact
from src.iam.models import Group
@ -42,9 +43,9 @@ async def get_org_by_id(db: db_dependency, org_model: org_model_dependency, org_
response = {
"name": org_model.name,
"status": org_model.status,
"owner_contact": (db.query(Contact).filter(Contact.id == org_model.owner_contact_id).first()),
"billing_contact": (db.query(Contact).filter(Contact.id == org_model.billing_contact_id).first()),
"security_contact": (db.query(Contact).filter(Contact.id == org_model.security_contact_id).first()),
"owner_contact": org_model.owner_contact_rel.email,
"billing_contact": org_model.billing_contact_rel.email,
"security_contact": org_model.security_contact_rel.email,
"root_user": org_model.root_user_email
}
@ -54,11 +55,17 @@ async def get_org_by_id(db: db_dependency, org_model: org_model_dependency, org_
@router.post("/")
async def create_org(db: db_dependency, org_request: OrgOrgPostRequest):
# TODO: Root user from current user
org_model = Org(**org_request.model_dump())
org_model = Org(name=org_request.name, intake_questionnaire=org_request.intake_questionnaire)
org_model.status = "partial" # Status is always set to partial at first, see update_questionnaire() doc
db.add(org_model)
db.flush()
for contact_type in ["billing_contact_id", "security_contact_id", "owner_contact_id"]:
contact_model = Contact(org_id=org_model.id)
db.add(contact_model)
db.flush()
org_model.__setattr__(contact_type, contact_model.id)
db.commit()
@ -95,26 +102,6 @@ async def update_status(db: db_dependency, status_request: OrgStatusPatchRequest
db.commit()
@router.patch("/{org_id}/contact")
async def update_contact(db: db_dependency, contact_request: OrgContactPatchRequest, org_id: Annotated[int, Path(gt=0)]):
org_model = db.query(Org).filter(Org.id == org_id).first()
if org_model is None:
raise HTTPException(status_code=404, detail="Organisation not found")
match contact_request.contact_type:
case "billing":
org_model.billing_contact_id = contact_request.contact_id
case "security":
org_model.security_contact_id = contact_request.contact_id
case "owner":
org_model.owner_contact_id = contact_request.contact_id
case _:
raise HTTPException(status_code=422, detail="Invalid contact type")
db.add(org_model)
db.commit()
@router.get("/{org_id}/users", response_model=list[OrgUserGetResponse])
async def get_users(db: db_dependency, org_id: Annotated[int, Path(gt=0)]):
org_exists = db.query(exists().where(Org.id == org_id)).scalar()
@ -147,29 +134,6 @@ async def delete_organisation_by_id(db: db_dependency, org_id: Annotated[int, Pa
db.commit()
@router.get("/{org_id}/contact/{contact_type}", response_model=OrgContactGetResponse)
async def get_contact(db: db_dependency, contact_type: ContactType, org_id: Annotated[int, Path(gt=0)]):
org_model = db.query(Org).filter(Org.id == org_id).first()
if org_model is None:
raise HTTPException(status_code=404, detail="Organisation not found")
match contact_type:
case "billing":
contact_id = org_model.billing_contact_id
case "security":
contact_id = org_model.security_contact_id
case "owner":
contact_id = org_model.owner_contact_id
case _:
raise HTTPException(status_code=422, detail="Invalid contact type")
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
return contact_model
@router.patch("/{org_id}/root_user")
async def update_root_user(db: db_dependency, org_model: org_model_dependency, org_id: Annotated[int, Path(gt=0)], root_user: Annotated[int, Query(gt=0)]):
# TODO: Request model, ditch query
@ -197,3 +161,56 @@ async def remove_user_from_org(db: db_dependency, org_model: org_model_dependenc
db.delete(orguser_model)
db.commit()
pass
@router.get("/{org_id}/contact", response_model=OrgContactGetResponse)
async def get_contact(db: db_dependency, org_model: org_model_dependency, contact_type: Annotated[ContactType, Query()], org_id: Annotated[int, Path(gt=0)]):
match contact_type:
case "billing":
contact_id = org_model.billing_contact_id
case "security":
contact_id = org_model.security_contact_id
case "owner":
contact_id = org_model.owner_contact_id
case _:
raise HTTPException(status_code=422, detail="Invalid contact type")
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
address = ContactAddress.model_validate(contact_model)
response = OrgContactGetResponse.model_construct(
**contact_model.__dict__,
address=address
)
return response
@router.patch("/{org_id}/contact")
async def update_contact(db: db_dependency, org_model: org_model_dependency, contact_type: Annotated[ContactType, Query()], contact_request: OrgContactPatchRequest, org_id: Annotated[int, Path(gt=0)]):
match contact_type:
case "billing":
contact_id = org_model.billing_contact_id
case "security":
contact_id = org_model.security_contact_id
case "owner":
contact_id = org_model.owner_contact_id
case _:
raise HTTPException(status_code=422, detail="Invalid contact type")
contact_model = (db.query(Contact).filter(Contact.id == contact_id).first())
if contact_model is None:
raise HTTPException(status_code=404, detail="Contact not found")
update_data = contact_request.model_dump(exclude_none=True)
for key, value in update_data.items():
if hasattr(contact_model, key):
setattr(contact_model, key, value)
else:
raise HTTPException(status_code=422, detail="Invalid keys in update request")
db.add(org_model)
db.commit()

40
src/organisation/schemas.py
View file

@ -7,9 +7,11 @@ Models:
"""
from typing import Optional
from pydantic import EmailStr, ConfigDict
from src.schemas import CustomBaseModel
from src.organisation.constants import Status, ContactType
from src.contact.schemas import ContactAddress
class OrgQuestionnaire(CustomBaseModel):
question_one: str
@ -21,10 +23,6 @@ class OrgOrgPostRequest(CustomBaseModel):
name: str
intake_questionnaire: Optional[OrgQuestionnaire] = None
billing_contact_id: Optional[int] = None
security_contact_id: Optional[int] = None
owner_contact_id: Optional[int] = None
class OrgQuestionnairePatchRequest(CustomBaseModel):
intake_questionnaire: OrgQuestionnaire
partial: bool
@ -33,8 +31,18 @@ class OrgStatusPatchRequest(CustomBaseModel):
status: Status
class OrgContactPatchRequest(CustomBaseModel):
contact_id: int
contact_type: ContactType
email: Optional[EmailStr] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
phonenumber: Optional[str] = None
vat_number: Optional[str] = None
post_office_box_number: Optional[str] = None
street_address: Optional[str] = None
street_address_line_2: Optional[str] = None
locality: Optional[str] = None
address_region: Optional[str] = None
country_code: Optional[str] = None
postal_code: Optional[str] = None
class OrgUserPostRequest(CustomBaseModel):
user_id: int
@ -43,16 +51,20 @@ class OrgUserGetResponse(CustomBaseModel):
user_id: int
class OrgContactGetResponse(CustomBaseModel):
email: str
first_name: str
last_name: str
phonenumber: str
model_config = ConfigDict(from_attributes=True, extra="ignore")
email: Optional[str] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
phonenumber: Optional[str] = None
vat_number: Optional[str] = None
address: ContactAddress
class OrgOrgGetResponse(CustomBaseModel):
name: str
status: Status
root_user: Optional[str] = None
owner_contact: Optional[OrgContactGetResponse] = None
billing_contact: Optional[OrgContactGetResponse] = None
security_contact: Optional[OrgContactGetResponse] = None
owner_contact: Optional[str] = None
billing_contact: Optional[str] = None
security_contact: Optional[str] = None