aaredb/backend/main.py

292 lines
9.9 KiB
Python

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi import HTTPException, status
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: int
firstname: str
lastname: str
phone_number: str
email: str
class Address(BaseModel):
id: int
street: str
city: str
zipcode: str
country: str
class Proposal(BaseModel):
id: 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
shippingStatus: str
arrivalStatus: str
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')
]
# 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',
ready_date='2023-09-30',
shipping_date='2023-10-01',
arrival_date='2023-10-02',
shippingStatus='Shipped',
arrivalStatus='Arrived',
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 Transit',
ready_date='2023-10-01',
shipping_date='2023-10-02',
arrival_date='2023-10-04',
shippingStatus='In Transit',
arrivalStatus='Pending',
qrcode='QR123DEWAR002'
),
Dewar(
id='DEWAR003',
dewar_name='Dewar Three',
tracking_number='TRACK125',
number_of_pucks=4,
number_of_samples=47,
return_address=[return_addresses[0]],
contact_person=[contacts[2]],
status='Pending',
shippingStatus='Ready for Shipping',
arrivalStatus='Pending',
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'] # 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]
# 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
)
]
@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():
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.get("/dewars", response_model=List[Dewar])
async def get_dewars():
return dewars
@app.post("/dewars", response_model=List[Dewar], status_code=status.HTTP_201_CREATED)
async def create_dewar(shipment: Dewar):
dewar_id = f'SHIP-{uuid.uuid4().hex[:8].upper()}' # Generates a unique shipment ID
shipment.id = dewar_id # Set the generated ID on the shipment object
dewars.append(shipment) # Add the modified shipment object to the list
return dewars # Return the list of all dewars
# Endpoint to get the number of dewars in each shipment
@app.get("/shipment_dewars")
async def get_shipment_dewars():
return [{"shipment_id": shipment.shipment_id, "number_of_dewars": shipment.get_number_of_dewars()} for shipment in
shipments]
@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):
logging.info(f"Received contact creation request: {contact}")
# 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="Contact with this email already exists."
)
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):
logging.info(f"Received contact creation request: {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."
)
return_addresses.append(address)
return address