from fastapi import FastAPI, HTTPException, status, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.logger import logger from pydantic import BaseModel from typing import List, Optional import logging import uuid logging.basicConfig(level=logging.INFO) app = FastAPI() app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) class ContactPerson(BaseModel): id: Optional[int] = None # Make id optional firstname: str lastname: str phone_number: str email: str class Address(BaseModel): id: Optional[int] = None # Make id optional street: str city: str zipcode: str country: str class Proposal(BaseModel): id: Optional[int] number: str class Dewar(BaseModel): id: Optional[str] = None dewar_name: str tracking_number: Optional[str] = None number_of_pucks: int number_of_samples: int return_address: List[Address] contact_person: List[ContactPerson] status: str ready_date: Optional[str] = None shipping_date: Optional[str] = None arrival_date: Optional[str] = None returning_date: Optional[str] = None qrcode: str class Shipment(BaseModel): shipment_id: Optional[str] = None shipment_name: str shipment_date: str shipment_status: str contact_person: List[ContactPerson] proposal_number: List[Proposal] return_address: List[Address] comments: Optional[str] = None dewars: List[Dewar] def get_number_of_dewars(self) -> int: return len(self.dewars) def get_shipment_contact_persons(self) -> List[ContactPerson]: return self.contact_person def get_shipment_return_addresses(self) -> List[Address]: return self.return_address def get_proposals(self) -> List[Proposal]: return self.proposal_number class Config: orm_mode = True # Example data for contacts contacts = [ ContactPerson(id=1, firstname="Frodo", lastname="Baggins", phone_number="123-456-7890", email="frodo.baggins@lotr.com"), ContactPerson(id=2, firstname="Samwise", lastname="Gamgee", phone_number="987-654-3210", email="samwise.gamgee@lotr.com"), ContactPerson(id=3, firstname="Aragorn", lastname="Elessar", phone_number="123-333-4444", email="aragorn.elessar@lotr.com"), ContactPerson(id=4, firstname="Legolas", lastname="Greenleaf", phone_number="555-666-7777", email="legolas.greenleaf@lotr.com"), ContactPerson(id=5, firstname="Gimli", lastname="Son of Gloin", phone_number="888-999-0000", email="gimli.sonofgloin@lotr.com"), ContactPerson(id=6, firstname="Gandalf", lastname="The Grey", phone_number="222-333-4444", email="gandalf.thegrey@lotr.com"), ContactPerson(id=7, firstname="Boromir", lastname="Son of Denethor", phone_number="111-222-3333", email="boromir.sonofdenethor@lotr.com"), ContactPerson(id=8, firstname="Galadriel", lastname="Lady of Lothlórien", phone_number="444-555-6666", email="galadriel.lothlorien@lotr.com"), ContactPerson(id=9, firstname="Elrond", lastname="Half-elven", phone_number="777-888-9999", email="elrond.halfelven@lotr.com"), ContactPerson(id=10, firstname="Eowyn", lastname="Shieldmaiden of Rohan", phone_number="000-111-2222", email="eowyn.rohan@lotr.com"), ] # Example data for return addresses return_addresses = [ Address(id=1, street='123 Hobbiton St', city='Shire', zipcode='12345', country='Middle Earth'), Address(id=2, street='456 Rohan Rd', city='Edoras', zipcode='67890', country='Middle Earth') Address(id=3, street='789 Greenwood Dr', city='Mirkwood', zipcode='13579', country='Middle Earth'), Address(id=4, street='321 Gondor Ave', city='Minas Tirith', zipcode='24680', country='Middle Earth'), Address(id=5, street='654 Falgorn Pass', city='Rivendell', zipcode='11223', country='Middle Earth') ] # Example data for dewars dewars = [ Dewar( id='DEWAR001', dewar_name='Dewar One', tracking_number='TRACK123', number_of_pucks=7, number_of_samples=70, return_address=[return_addresses[0]], contact_person=[contacts[0]], status='Ready for Shipping', ready_date='2023-09-30', shipping_date='', arrival_date='', returning_date='', qrcode='QR123DEWAR001' ), Dewar( id='DEWAR002', dewar_name='Dewar Two', tracking_number='TRACK124', number_of_pucks=3, number_of_samples=33, return_address=[return_addresses[1]], contact_person=[contacts[1]], status='In Preparation', ready_date='', shipping_date='', arrival_date='', returning_date='', qrcode='QR123DEWAR002' ), Dewar( id='DEWAR003', dewar_name='Dewar Three', tracking_number='TRACK125', number_of_pucks=7, number_of_samples=72, return_address=[return_addresses[0]], contact_person=[contacts[2]], status='Not Shipped', ready_date='2024.01.01', shipping_date='', arrival_date='', returning_date='', qrcode='QR123DEWAR003' ), Dewar( id='DEWAR004', dewar_name='Dewar Four', tracking_number='', number_of_pucks=7, number_of_samples=70, return_address=[return_addresses[0]], contact_person=[contacts[2]], status='Delayed', ready_date='2024.01.01', shipping_date='2024.01.02', arrival_date='', returning_date='', qrcode='QR123DEWAR003' ), Dewar( id='DEWAR005', dewar_name='Dewar Five', tracking_number='', number_of_pucks=3, number_of_samples=30, return_address=[return_addresses[0]], contact_person=[contacts[2]], status='Returned', ready_date='2024.01.01', shipping_date='2024.01.02', arrival_date='2024.01.03', returning_date='2024.01.07', qrcode='QR123DEWAR003' ), ] # Proposal data inspired by the Lord of the Rings proposals = [ Proposal(id=1, number="PROPOSAL-FRODO-001"), # "The Quest for the Ring" Proposal(id=2, number="PROPOSAL-GANDALF-002"), # "The Fellowship's Journey" Proposal(id=3, number="PROPOSAL-ARAGORN-003"), # "Return of the King" Proposal(id=4, number="PROPOSAL-SAURON-004"), # "The Dark Lord's Plot" Proposal(id=5, number="PROPOSAL-MORDOR-005"), # "The Road to Mount Doom" ] # Example: Attach specific Dewars by their ids to shipments specific_dewar_ids1 = ['DEWAR003'] # The IDs of the Dewars you want to attach to the first shipment specific_dewar_ids2 = ['DEWAR001', 'DEWAR002'] specific_dewar_ids3 = ['DEWAR003', 'DEWAR004', 'DEWAR005'] # The IDs of the Dewars you want to attach to the second shipment # Find the Dewars with the matching ids specific_dewars1 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids1] specific_dewars2 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids2] specific_dewars3 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids3] # Define shipments with the selected Dewars shipments = [ Shipment( shipment_id='SHIPMORDOR', shipment_date='2024-10-10', shipment_name='Shipment from Mordor', shipment_status='Delivered', contact_person=[contacts[1]], proposal_number=[proposals[1]], return_address=[return_addresses[0]], comments='Handle with care', dewars=specific_dewars1 # Attach specific Dewars for this shipment ), Shipment( shipment_id='SHIPMORDOR2', shipment_date='2024-10-24', shipment_name='Shipment from Mordor', shipment_status='In Transit', contact_person=[contacts[3]], proposal_number=[proposals[2]], return_address=[return_addresses[1]], # Changed index to a valid one comments='Contains the one ring', dewars=specific_dewars2 # Attach specific Dewars for this shipment ), Shipment( shipment_id='SHIPMORDOR3', shipment_date='2024-10-28', shipment_name='Shipment from Mordor', shipment_status='In Transit', contact_person=[contacts[4]], proposal_number=[proposals[3]], return_address=[return_addresses[0]], # Changed index to a valid one comments='Contains the one ring', dewars=specific_dewars3 ) ] @app.get("/contacts", response_model=List[ContactPerson]) async def get_contacts(): return contacts @app.get("/return_addresses", response_model=List[Address]) async def get_return_addresses(): return return_addresses @app.get("/proposals", response_model=List[Proposal]) async def get_proposals(): return proposals @app.get("/shipments", response_model=List[Shipment]) async def get_shipments(shipment_id: Optional[str] = Query(None, description="ID of the specific shipment to retrieve")): if shipment_id: shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None) if not shipment: raise HTTPException(status_code=404, detail="Shipment not found") return [shipment] return shipments @app.delete("/shipments/{shipment_id}", status_code=status.HTTP_204_NO_CONTENT) async def delete_shipment(shipment_id: str): global shipments # Use global variable to access the shipments list shipments = [shipment for shipment in shipments if shipment.shipment_id != shipment_id] @app.post("/shipments/{shipment_id}/add_dewar", response_model=Shipment) async def add_dewar_to_shipment(shipment_id: str, dewar_id: str): # Find the shipment by id shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None) if not shipment: raise HTTPException(status_code=404, detail="Shipment not found") # Find the dewar by id dewar = next((dw for dw in dewars if dw.id == dewar_id), None) if not dewar: raise HTTPException(status_code=404, detail="Dewar not found") # Add the dewar to the shipment if dewar not in shipment.dewars: shipment.dewars.append(dewar) return shipment @app.put("/shipments/{shipment_id}", response_model=Shipment) async def update_shipment(shipment_id: str, updated_shipment: Shipment): global shipments shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None) if not shipment: raise HTTPException(status_code=404, detail="Shipment not found") logger.info(f"Updating shipment: {shipment_id}") logger.info(f"Updated shipment data: {updated_shipment}") # Create a dictionary of existing dewars for fast lookup existing_dewar_dict = {dewar.id: dewar for dewar in shipment.dewars} # Update or add dewars from the updated shipment data for updated_dewar in updated_shipment.dewars: if updated_dewar.id in existing_dewar_dict: # Update existing dewar existing_dewar_dict[updated_dewar.id].dewar_name = updated_dewar.dewar_name existing_dewar_dict[updated_dewar.id].tracking_number = updated_dewar.tracking_number existing_dewar_dict[updated_dewar.id].number_of_pucks = updated_dewar.number_of_pucks existing_dewar_dict[updated_dewar.id].number_of_samples = updated_dewar.number_of_samples existing_dewar_dict[updated_dewar.id].return_address = updated_dewar.return_address existing_dewar_dict[updated_dewar.id].contact_person = updated_dewar.contact_person existing_dewar_dict[updated_dewar.id].status = updated_dewar.status existing_dewar_dict[updated_dewar.id].ready_date = updated_dewar.ready_date existing_dewar_dict[updated_dewar.id].shipping_date = updated_dewar.shipping_date existing_dewar_dict[updated_dewar.id].arrival_date = updated_dewar.arrival_date existing_dewar_dict[updated_dewar.id].returning_date = updated_dewar.returning_date existing_dewar_dict[updated_dewar.id].qrcode = updated_dewar.qrcode else: # Add new dewar shipment.dewars.append(updated_dewar) # Update the shipment's fields shipment.shipment_name = updated_shipment.shipment_name shipment.shipment_date = updated_shipment.shipment_date shipment.shipment_status = updated_shipment.shipment_status shipment.contact_person = updated_shipment.contact_person shipment.proposal_number = updated_shipment.proposal_number shipment.return_address = updated_shipment.return_address shipment.comments = updated_shipment.comments logger.info(f"Shipment after update: {shipment}") return shipment @app.get("/dewars", response_model=List[Dewar]) async def get_dewars(): return dewars @app.post("/dewars", response_model=Dewar, status_code=status.HTTP_201_CREATED) async def create_dewar(dewar: Dewar) -> Dewar: dewar_id = f'DEWAR-{uuid.uuid4().hex[:8].upper()}' # Generates a unique dewar ID dewar.id = dewar_id # Set the generated ID on the dewar object dewars.append(dewar) # Add the modified dewar object to the list return dewar # Return the newly created dewar @app.delete("/shipments/{shipment_id}/remove_dewar/{dewar_id}", response_model=Shipment) async def remove_dewar_from_shipment(shipment_id: str, dewar_id: str): """Remove a dewar from a shipment.""" # Find the shipment by ID shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None) if not shipment: raise HTTPException(status_code=404, detail="Shipment not found") # Remove the dewar from the shipment shipment.dewars = [dw for dw in shipment.dewars if dw.id != dewar_id] return shipment @app.get("/shipment_contact_persons") async def get_shipment_contact_persons(): return [{"shipment_id": shipment.shipment_id, "contact_person": shipment.get_shipment_contact_persons()} for shipment in shipments] @app.post("/shipments", response_model=Shipment, status_code=status.HTTP_201_CREATED) async def create_shipment(shipment: Shipment): # Automatically generate a shipment ID shipment_id = f'SHIP-{uuid.uuid4().hex[:8].upper()}' # Generates a unique shipment ID shipment.shipment_id = shipment_id # Set the generated ID # Append the shipment to the list shipments.append(shipment) return shipment # Creation of a new contact @app.post("/contacts", response_model=ContactPerson, status_code=status.HTTP_201_CREATED) async def create_contact(contact: ContactPerson): # Check for duplicate contact by email (or other unique fields) if any(c.email == contact.email for c in contacts): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="This contact already exists." ) # Find the next available id if contacts: max_id = max(c.id for c in contacts) contact.id = max_id + 1 if contact.id is None else contact.id else: contact.id = 1 if contact.id is None else contact.id contacts.append(contact) return contact # Creation of a return address @app.post("/return_addresses", response_model=Address, status_code=status.HTTP_201_CREATED) async def create_return_address(address: Address): # Check for duplicate address by city if any(a.city == address.city for a in return_addresses): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Address in this city already exists." ) # Find the next available id if return_addresses: max_id = max(a.id for a in return_addresses) address.id = max_id + 1 if address.id is None else address.id else: address.id = 1 if address.id is None else address.id return_addresses.append(address) return address