gotthard_g/aaredb
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Better integration of sqlite3 database

Browse Source
This commit is contained in:
GotthardG
2024-11-02 00:54:37 +01:00
parent 48cd233231
commit a01114a178
18 changed files with 835 additions and 582 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
backend/app/crud.py Normal file
View File

@ -0,0 +1,20 @@
from sqlalchemy.orm import Session, joinedload
def get_shipments(db: Session):
from app.models import Shipment
return db.query(Shipment).options(
joinedload(Shipment.contact_person),
joinedload(Shipment.return_address),
joinedload(Shipment.proposal),
joinedload(Shipment.dewars)
).all()
def get_shipment_by_id(db: Session, shipment_id: str):
from app.models import Shipment
shipment = db.query(Shipment).options(
joinedload(Shipment.contact_person),
joinedload(Shipment.return_address),
joinedload(Shipment.proposal),
joinedload(Shipment.dewars)
).filter(Shipment.shipment_id == shipment_id).first()
return shipment

97
backend/app/data/data.py
View File

@ -1,62 +1,75 @@
from typing import List
from app.models import ContactPerson, Address, Dewar, Proposal, Shipment
from datetime import datetime
contacts: List[ContactPerson] = [
ContactPerson(id=1, firstname="Frodo", lastname="Baggins", phone_number="123-456-7890", email="frodo.baggins@lotr.com"),
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"),
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"),
]
return_addresses: List[Address] = [
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')
Address(id=5, street='654 Falgorn Pass', city='Rivendell', zipcode='11223', country='Middle Earth'),
]
dewars: List[Dewar] = [
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'
return_address_id=1, contact_person_id=1, status='Ready for Shipping',
ready_date=datetime.strptime('2023-09-30', '%Y-%m-%d'), shipping_date=None, arrival_date=None,
returning_date=None, 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'
return_address_id=2, contact_person_id=2, status='In Preparation',
ready_date=None, shipping_date=None, arrival_date=None, returning_date=None, 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'
return_address_id=1, contact_person_id=3, status='Not Shipped',
ready_date=datetime.strptime('2024-01-01', '%Y-%m-%d'), shipping_date=None, arrival_date=None,
returning_date=None, 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'
return_address_id=1, contact_person_id=3, status='Delayed',
ready_date=datetime.strptime('2024-01-01', '%Y-%m-%d'), shipping_date=datetime.strptime('2024-01-02', '%Y-%m-%d'),
arrival_date=None, returning_date=None, qrcode='QR123DEWAR004',
),
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'
return_address_id=1, contact_person_id=3, status='Returned',
arrival_date=datetime.strptime('2024-01-03', '%Y-%m-%d'), returning_date=datetime.strptime('2024-01-07', '%Y-%m-%d'),
qrcode='QR123DEWAR005',
),
]
proposals: List[Proposal] = [
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"
proposals = [
Proposal(id=1, number="PROPOSAL-FRODO-001"),
Proposal(id=2, number="PROPOSAL-GANDALF-002"),
Proposal(id=3, number="PROPOSAL-ARAGORN-003"),
Proposal(id=4, number="PROPOSAL-SAURON-004"),
Proposal(id=5, number="PROPOSAL-MORDOR-005"),
]
specific_dewar_ids1 = ['DEWAR003']
@ -69,18 +82,18 @@ specific_dewars3 = [dewar for dewar in dewars if dewar.id in specific_dewar_ids3
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
shipment_id="SHIPMENT001", shipment_date=datetime.strptime('2024-10-10', '%Y-%m-%d'),
shipment_name='Shipment from Mordor', shipment_status='Delivered', contact_person_id=2,
proposal_id=3, return_address_id=1, comments='Handle with care', dewars=specific_dewars1
),
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]], comments='Contains the one ring', dewars=specific_dewars2
shipment_id="SHIPMENT002", shipment_date=datetime.strptime('2024-10-24', '%Y-%m-%d'),
shipment_name='Shipment from Mordor', shipment_status='In Transit', contact_person_id=4,
proposal_id=4, return_address_id=2, comments='Contains the one ring', dewars=specific_dewars2
),
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]], comments='Contains the one ring', dewars=specific_dewars3
)
shipment_id="SHIPMENT003", shipment_date=datetime.strptime('2024-10-28', '%Y-%m-%d'),
shipment_name='Shipment from Mordor', shipment_status='In Transit', contact_person_id=5,
proposal_id=5, return_address_id=1, comments='Contains the one ring', dewars=specific_dewars3
),
]

27
backend/app/database.py
View File

@ -1,15 +1,16 @@
from sqlalchemy.orm import Session
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
DATABASE_URL = "sqlite:///./test.db"
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
# Database setup
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency
def get_db():
db = SessionLocal()
@ -18,7 +19,21 @@ def get_db():
finally:
db.close()
# Initialize the database
def init_db():
import app.models # Import all models here for metadata.create_all() to recognize them
Base.metadata.create_all(bind=engine)
# Import inside function to avoid circular dependency
from app import models
Base.metadata.create_all(bind=engine)
def load_sample_data(session: Session):
# Import inside function to avoid circular dependency
from app.data import contacts, return_addresses, dewars, proposals, shipments
from app import models # Ensure these imports are correct
if session.query(models.ContactPerson).first():
return
session.add_all(contacts + return_addresses + dewars + proposals + shipments)
session.commit()

9
backend/app/dependencies.py Normal file
View File

@ -0,0 +1,9 @@
# app/dependencies.py
from app.database import SessionLocal # Import SessionLocal from database.py
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()

7
backend/app/init_db.py Normal file
View File

@ -0,0 +1,7 @@
from app.database import init_db
def initialize_database():
init_db()
if __name__ == "__main__":
initialize_database()

22
backend/app/main.py
View File

@ -1,21 +1,37 @@
# app/main.py
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.routers import address, contact, proposal, dewar, shipment
from app.database import Base, engine, SessionLocal, load_sample_data
app = FastAPI()
# Apply CORS middleware
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # Enable CORS for all origins for now
allow_origins=["*"], # Enable CORS for all origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Include your routers
@app.on_event("startup")
def on_startup():
# Drop and recreate database schema
Base.metadata.drop_all(bind=engine)
Base.metadata.create_all(bind=engine)
db = SessionLocal()
try:
load_sample_data(db)
finally:
db.close()
# Include routers with correct configuration
app.include_router(contact.router, prefix="/contacts", tags=["contacts"])
app.include_router(address.router, prefix="/return_addresses", tags=["return_addresses"])
app.include_router(address.router, prefix="/addresses", tags=["addresses"])
app.include_router(proposal.router, prefix="/proposals", tags=["proposals"])
app.include_router(dewar.router, prefix="/dewars", tags=["dewars"])
app.include_router(shipment.router, prefix="/shipments", tags=["shipments"])

119
backend/app/models.py
View File

@ -1,61 +1,78 @@
from pydantic import BaseModel
from typing import List, Optional
from sqlalchemy import Column, Integer, String, Date, ForeignKey
from sqlalchemy.orm import relationship
from app.database import Base # Ensure this imports correctly
class ContactPerson(BaseModel):
id: Optional[int] = None
firstname: str
lastname: str
phone_number: str
email: str
class Address(BaseModel):
id: Optional[int] = None
street: str
city: str
zipcode: str
country: str
# SQLAlchemy ORM models
class Shipment(Base):
__tablename__ = "shipments"
class Proposal(BaseModel):
id: Optional[int] = None
number: str
shipment_id = Column(String, primary_key=True, index=True)
shipment_name = Column(String, index=True)
shipment_date = Column(Date)
shipment_status = Column(String)
comments = Column(String, nullable=True)
contact_person_id = Column(Integer, ForeignKey("contact_persons.id"))
return_address_id = Column(Integer, ForeignKey("addresses.id"))
proposal_id = Column(Integer, ForeignKey("proposals.id"))
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
contact_person = relationship("ContactPerson", back_populates="shipments")
return_address = relationship("Address", back_populates="shipments")
proposal = relationship("Proposal", back_populates="shipments")
dewars = relationship("Dewar", back_populates="shipment")
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)
class ContactPerson(Base):
__tablename__ = "contact_persons"
def get_shipment_contact_persons(self) -> List[ContactPerson]:
return self.contact_person
id = Column(Integer, primary_key=True, index=True)
firstname = Column(String)
lastname = Column(String)
phone_number = Column(String)
email = Column(String)
def get_shipment_return_addresses(self) -> List[Address]:
return self.return_address
shipments = relationship("Shipment", back_populates="contact_person")
def get_proposals(self) -> List[Proposal]:
return self.proposal_number
class Config:
from_attributes = True
class Address(Base):
__tablename__ = "addresses"
id = Column(Integer, primary_key=True, index=True)
street = Column(String)
city = Column(String)
zipcode = Column(String)
country = Column(String)
shipments = relationship("Shipment", back_populates="return_address")
class Dewar(Base):
__tablename__ = "dewars"
id = Column(String, primary_key=True, index=True)
dewar_name = Column(String)
tracking_number = Column(String)
number_of_pucks = Column(Integer)
number_of_samples = Column(Integer)
status = Column(String)
ready_date = Column(Date, nullable=True)
shipping_date = Column(Date, nullable=True)
arrival_date = Column(Date, nullable=True)
returning_date = Column(Date, nullable=True)
qrcode = Column(String)
shipment_id = Column(String, ForeignKey("shipments.shipment_id"))
return_address_id = Column(Integer, ForeignKey("addresses.id")) # Added
contact_person_id = Column(Integer, ForeignKey("contact_persons.id")) # Added
shipment = relationship("Shipment", back_populates="dewars")
return_address = relationship("Address") # Defines relationship with Address
contact_person = relationship("ContactPerson") # Defines relationship with ContactPerson
class Proposal(Base):
__tablename__ = "proposals"
id = Column(Integer, primary_key=True, index=True)
number = Column(String)
shipments = relationship("Shipment", back_populates="proposal")

42
backend/app/routers/address.py
View File

@ -1,26 +1,32 @@
from fastapi import APIRouter, HTTPException, status
from typing import List, Optional
from app.data.data import return_addresses
from app.models import Address # Import the Address model
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.orm import Session
from typing import List
from app.schemas import Address as AddressSchema, AddressCreate
from app.models import Address as AddressModel
from app.dependencies import get_db
router = APIRouter()
@router.get("/", response_model=List[Address])
async def get_return_addresses():
return return_addresses
@router.get("/", response_model=List[AddressSchema])
async def get_return_addresses(db: Session = Depends(get_db)):
return db.query(AddressModel).all()
@router.post("/", response_model=Address, status_code=status.HTTP_201_CREATED)
async def create_return_address(address: Address):
if any(a.city == address.city for a in return_addresses):
@router.post("/", response_model=AddressSchema, status_code=status.HTTP_201_CREATED)
async def create_return_address(address: AddressCreate, db: Session = Depends(get_db)):
if db.query(AddressModel).filter(AddressModel.city == address.city).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Address in this city already exists."
)
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
db_address = AddressModel(
street=address.street,
city=address.city,
zipcode=address.zipcode,
country=address.country
)
db.add(db_address)
db.commit()
db.refresh(db_address)
return db_address

35
backend/app/routers/contact.py
View File

@ -1,26 +1,31 @@
# app/routers/contact.py
from fastapi import APIRouter, HTTPException, status
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.orm import Session
from typing import List
from app.data.data import contacts
from app.models import ContactPerson
from app.schemas import ContactPerson, ContactPersonCreate
from app.models import ContactPerson as ContactPersonModel
from app.dependencies import get_db
router = APIRouter()
@router.get("/", response_model=List[ContactPerson])
async def get_contacts():
return contacts
async def get_contacts(db: Session = Depends(get_db)):
return db.query(ContactPersonModel).all()
@router.post("/", response_model=ContactPerson, status_code=status.HTTP_201_CREATED)
async def create_contact(contact: ContactPerson):
if any(c.email == contact.email for c in contacts):
async def create_contact(contact: ContactPersonCreate, db: Session = Depends(get_db)):
if db.query(ContactPersonModel).filter(ContactPersonModel.email == contact.email).first():
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="This contact already exists."
)
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
db_contact = ContactPersonModel(
firstname=contact.firstname,
lastname=contact.lastname,
phone_number=contact.phone_number,
email=contact.email
)
db.add(db_contact)
db.commit()
db.refresh(db_contact)
return db_contact

45
backend/app/routers/dewar.py
View File

@ -1,19 +1,38 @@
from fastapi import APIRouter, HTTPException, status
from typing import List, Optional
from fastapi import APIRouter, HTTPException, status, Depends
from sqlalchemy.orm import Session
from typing import List
import uuid
from app.data.data import dewars, contacts, return_addresses
from app.models import Dewar, ContactPerson, Address
from app.schemas import Dewar as DewarSchema, DewarCreate
from app.models import Dewar as DewarModel
from app.dependencies import get_db
router = APIRouter()
@router.get("/", response_model=List[Dewar])
async def get_dewars():
return dewars
@router.get("/", response_model=List[DewarSchema])
async def get_dewars(db: Session = Depends(get_db)):
return db.query(DewarModel).all()
@router.post("/", response_model=Dewar, status_code=status.HTTP_201_CREATED)
async def create_dewar(dewar: Dewar) -> Dewar:
@router.post("/", response_model=DewarSchema, status_code=status.HTTP_201_CREATED)
async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> DewarSchema:
dewar_id = f'DEWAR-{uuid.uuid4().hex[:8].upper()}'
dewar.id = dewar_id
dewars.append(dewar)
return dewar
db_dewar = DewarModel(
id=dewar_id,
dewar_name=dewar.dewar_name,
tracking_number=dewar.tracking_number,
number_of_pucks=dewar.number_of_pucks,
number_of_samples=dewar.number_of_samples,
status=dewar.status,
ready_date=dewar.ready_date,
shipping_date=dewar.shipping_date,
arrival_date=dewar.arrival_date,
returning_date=dewar.returning_date,
qrcode=dewar.qrcode,
contact_person_id=dewar.contact_person_id,
return_address_id=dewar.return_address_id
)
db.add(db_dewar)
db.commit()
db.refresh(db_dewar)
return db_dewar

18
backend/app/routers/proposal.py
View File

@ -1,11 +1,13 @@
from fastapi import APIRouter
from typing import List, Optional
from app.data.data import proposals
from app.models import Proposal # Import the Address model
# app/routers/proposal.py
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from typing import List
from app.schemas import Proposal as ProposalSchema
from app.models import Proposal as ProposalModel
from app.dependencies import get_db
router = APIRouter()
@router.get("/", response_model=List[Proposal])
async def get_proposals():
return proposals
@router.get("/", response_model=List[ProposalSchema])
async def get_proposals(db: Session = Depends(get_db)):
return db.query(ProposalModel).all()

189
backend/app/routers/shipment.py
View File

@ -1,85 +1,154 @@
from fastapi import APIRouter, HTTPException, status, Query
from fastapi import APIRouter, HTTPException, status, Query, Depends
from sqlalchemy.orm import Session
from typing import List, Optional
import uuid
from app.data.data import shipments, dewars
from app.models import Shipment, Dewar, ContactPerson, Address, Proposal
from app.schemas import ShipmentCreate, Shipment as ShipmentSchema, ContactPerson as ContactPersonSchema
from app.database import get_db
from app.crud import get_shipments, get_shipment_by_id
router = APIRouter()
@router.get("/", response_model=List[Shipment])
async def get_shipments(shipment_id: Optional[str] = Query(None, description="ID of the specific shipment to retrieve")):
@router.get("", response_model=List[ShipmentSchema])
async def fetch_shipments(shipment_id: Optional[str] = Query(None),
db: Session = Depends(get_db)):
if shipment_id:
shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None)
shipment = get_shipment_by_id(db, shipment_id)
if not shipment:
raise HTTPException(status_code=404, detail="Shipment not found")
return [shipment]
return shipments
return get_shipments(db)
@router.post("", response_model=ShipmentSchema, status_code=status.HTTP_201_CREATED)
async def create_shipment(shipment: ShipmentCreate, db: Session = Depends(get_db)):
from app.models import Shipment as ShipmentModel, ContactPerson as ContactPersonModel, Address as AddressModel, \
Proposal as ProposalModel, Dewar as DewarModel
contact_person = db.query(ContactPersonModel).filter(ContactPersonModel.id == shipment.contact_person_id).first()
return_address = db.query(AddressModel).filter(AddressModel.id == shipment.return_address_id).first()
proposal = db.query(ProposalModel).filter(ProposalModel.id == shipment.proposal_id).first()
if not (contact_person and return_address and proposal):
raise HTTPException(status_code=404, detail="Associated entity not found")
shipment_id = f'SHIP-{uuid.uuid4().hex[:8].upper()}'
db_shipment = ShipmentModel(
shipment_id=shipment_id,
shipment_name=shipment.shipment_name,
shipment_date=shipment.shipment_date,
shipment_status=shipment.shipment_status,
comments=shipment.comments,
contact_person_id=contact_person.id,
return_address_id=return_address.id,
proposal_id=proposal.id,
)
# Handling dewars association
if shipment.dewars:
dewars = db.query(DewarModel).filter(DewarModel.id.in_(shipment.dewars)).all()
if len(dewars) != len(shipment.dewars):
raise HTTPException(status_code=404, detail="One or more dewars not found")
db_shipment.dewars.extend(dewars)
db.add(db_shipment)
db.commit()
db.refresh(db_shipment)
return db_shipment
@router.delete("/{shipment_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_shipment(shipment_id: str):
global shipments
shipments = [shipment for shipment in shipments if shipment.shipment_id != shipment_id]
@router.post("/{shipment_id}/add_dewar", response_model=Shipment)
async def add_dewar_to_shipment(shipment_id: str, dewar_id: str):
shipment = next((sh for sh in shipments if sh.shipment_id == shipment_id), None)
async def delete_shipment(shipment_id: str, db: Session = Depends(get_db)):
from app.models import Shipment as ShipmentModel
shipment = db.query(ShipmentModel).filter(ShipmentModel.shipment_id == shipment_id).first()
if not shipment:
raise HTTPException(status_code=404, detail="Shipment not found")
dewar = next((dw for dw in dewars if dw.id == dewar_id), None)
db.delete(shipment)
db.commit()
return
@router.put("/shipments/{shipment_id}", response_model=ShipmentSchema)
async def update_shipment(shipment_id: str, updated_shipment: ShipmentCreate, db: Session = Depends(get_db)):
from app.models import Shipment as ShipmentModel, ContactPerson as ContactPersonModel, Address as AddressModel, \
Dewar as DewarModel
# Log incoming payload for detailed inspection
print("Received payload:", json.dumps(updated_shipment.dict(), indent=2))
try:
shipment = db.query(ShipmentModel).filter(ShipmentModel.shipment_id == shipment_id).first()
if not shipment:
raise HTTPException(status_code=404, detail="Shipment not found")
# Validate relationships by IDs
contact_person = db.query(ContactPersonModel).filter(
ContactPersonModel.id == updated_shipment.contact_person_id).first()
return_address = db.query(AddressModel).filter(AddressModel.id == updated_shipment.return_address_id).first()
if not contact_person:
raise HTTPException(status_code=404, detail="Contact person not found")
if not return_address:
raise HTTPException(status_code=404, detail="Return address not found")
# Handling dewars association by IDs
dewars_ids = [d['id'] for d in updated_shipment.dewars]
dewars = db.query(DewarModel).filter(DewarModel.id.in_(dewars_ids)).all()
if len(dewars) != len(dewars_ids):
raise HTTPException(status_code=422, detail="One or more dewars not found")
# Update shipment details
shipment.shipment_name = updated_shipment.shipment_name
shipment.shipment_date = updated_shipment.shipment_date
shipment.shipment_status = updated_shipment.shipment_status
shipment.comments = updated_shipment.comments
shipment.contact_person_id = updated_shipment.contact_person_id
shipment.return_address_id = updated_shipment.return_address_id
shipment.dewars = dewars
db.commit()
db.refresh(shipment)
return shipment
except Exception as e:
print(f"Update failed with exception: {e}")
raise HTTPException(status_code=500, detail="Internal server error")
@router.post("/{shipment_id}/add_dewar", response_model=ShipmentSchema)
async def add_dewar_to_shipment(shipment_id: str, dewar_id: str, db: Session = Depends(get_db)):
from app.models import Shipment as ShipmentModel, Dewar as DewarModel
shipment = db.query(ShipmentModel).filter(ShipmentModel.shipment_id == shipment_id).first()
if not shipment:
raise HTTPException(status_code=404, detail="Shipment not found")
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
if not dewar:
raise HTTPException(status_code=404, detail="Dewar not found")
if dewar not in shipment.dewars:
shipment.dewars.append(dewar)
db.commit()
db.refresh(shipment)
return shipment
@router.put("/{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)
@router.delete("/{shipment_id}/remove_dewar/{dewar_id}", response_model=ShipmentSchema)
async def remove_dewar_from_shipment(shipment_id: str, dewar_id: str, db: Session = Depends(get_db)):
from app.models import Shipment as ShipmentModel, Dewar as DewarModel
shipment = db.query(ShipmentModel).filter(ShipmentModel.shipment_id == shipment_id).first()
if not shipment:
raise HTTPException(status_code=404, detail="Shipment not found")
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
existing_dewar_dict = {dewar.id: dewar for dewar in shipment.dewars}
for updated_dewar in updated_shipment.dewars:
if updated_dewar.id in existing_dewar_dict:
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:
shipment.dewars.append(updated_dewar)
return shipment
@router.post("/", response_model=Shipment, status_code=status.HTTP_201_CREATED)
async def create_shipment(shipment: Shipment):
shipment_id = f'SHIP-{uuid.uuid4().hex[:8].upper()}'
shipment.shipment_id = shipment_id
shipments.append(shipment)
return shipment
@router.delete("/{shipment_id}/remove_dewar/{dewar_id}", response_model=Shipment)
async def remove_dewar_from_shipment(shipment_id: str, dewar_id: str):
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")
shipment.dewars = [dw for dw in shipment.dewars if dw.id != dewar_id]
db.commit()
db.refresh(shipment)
return shipment
@router.get("/contact_persons", response_model=List[ContactPerson])
async def get_shipment_contact_persons():
return [{"shipment_id": shipment.shipment_id, "contact_person": shipment.get_shipment_contact_persons()} for shipment in shipments]
@router.get("/contact_persons", response_model=List[ContactPersonSchema])
async def get_shipment_contact_persons(db: Session = Depends(get_db)):
from app.models import ContactPerson as ContactPersonModel
contact_persons = db.query(ContactPersonModel).all()
return contact_persons

116
backend/app/schemas.py
View File

@ -1,59 +1,117 @@
from pydantic import BaseModel
from typing import List, Optional
from pydantic import BaseModel, EmailStr
from datetime import date
class ContactPersonSchema(BaseModel):
id: Optional[int]
# Base class for Contact Person
class ContactPersonBase(BaseModel):
firstname: str
lastname: str
phone_number: str
email: str
email: EmailStr
# Create schema for Contact Person
class ContactPersonCreate(ContactPersonBase):
pass
# Response schema for Contact Person with ID
class ContactPerson(ContactPersonBase):
id: int
class Config:
from_attributes = True # Update here
from_attributes = True
class AddressSchema(BaseModel):
id: Optional[int]
# Create schema for Address
class AddressCreate(BaseModel):
street: str
city: str
zipcode: str
country: str
class Config:
from_attributes = True # Update here
class DewarSchema(BaseModel):
id: Optional[str]
# Response schema for Address with ID
class Address(AddressCreate):
id: int
class Config:
from_attributes = True
# Create schema for Dewar
class DewarCreate(BaseModel):
dewar_name: str
tracking_number: Optional[str]
tracking_number: str
number_of_pucks: int
number_of_samples: int
status: str
ready_date: Optional[str]
shipping_date: Optional[str]
arrival_date: Optional[str]
returning_date: Optional[str]
ready_date: Optional[date]
shipping_date: Optional[date]
arrival_date: Optional[date]
returning_date: Optional[date]
qrcode: str
contact_person_id: Optional[int]
return_address_id: Optional[int]
# Response schema for Dewar
class Dewar(BaseModel):
id: str
dewar_name: str
tracking_number: str
number_of_pucks: int
number_of_samples: int
status: str
ready_date: Optional[date]
shipping_date: Optional[date]
arrival_date: Optional[date]
returning_date: Optional[date]
qrcode: str
shipment_id: Optional[str]
contact_person: Optional[ContactPerson]
return_address: Optional[Address]
class Config:
from_attributes = True # Update here
from_attributes = True
class ProposalSchema(BaseModel):
id: Optional[int]
# Proposal schema
class Proposal(BaseModel):
id: int
number: str
class Config:
from_attributes = True # Update here
from_attributes = True
class ShipmentSchema(BaseModel):
shipment_id: Optional[str]
# Response schema for Shipment
class Shipment(BaseModel):
shipment_id: str
shipment_name: str
shipment_date: str
shipment_date: date
shipment_status: str
contact_person: List[ContactPersonSchema]
proposal_number: List[ProposalSchema]
return_address: List[AddressSchema]
comments: Optional[str] = None
dewars: List[DewarSchema]
comments: Optional[str]
contact_person: Optional[ContactPerson]
return_address: Optional[Address]
proposal: Optional[Proposal]
dewars: Optional[List[Dewar]] = []
class Config:
from_attributes = True # Update here
from_attributes = True
# Create schema for Shipment
class ShipmentCreate(BaseModel):
shipment_name: str
shipment_date: date
shipment_status: str
comments: Optional[str] = ""
contact_person_id: int
return_address_id: int
proposal_id: int # Change "proposal_number_id" to "proposal_id"
dewars: Optional[List[str]] = []
class Config:
from_attributes = True