Fix formatting with black
This commit is contained in:
@ -13,7 +13,7 @@ from app.schemas import (
|
||||
DewarTypeCreate,
|
||||
DewarSerialNumber as DewarSerialNumberSchema,
|
||||
DewarSerialNumberCreate,
|
||||
Shipment as ShipmentSchema # Clearer name for schema
|
||||
Shipment as ShipmentSchema, # Clearer name for schema
|
||||
)
|
||||
from app.models import (
|
||||
Dewar as DewarModel,
|
||||
@ -21,7 +21,7 @@ from app.models import (
|
||||
Sample as SampleModel,
|
||||
DewarType as DewarTypeModel,
|
||||
DewarSerialNumber as DewarSerialNumberModel,
|
||||
Shipment as ShipmentModel # Clearer name for model
|
||||
Shipment as ShipmentModel, # Clearer name for model
|
||||
)
|
||||
from app.dependencies import get_db
|
||||
import uuid
|
||||
@ -32,23 +32,32 @@ from PIL import ImageFont, ImageDraw, Image
|
||||
from reportlab.lib.pagesizes import A5, landscape
|
||||
from reportlab.lib.units import cm
|
||||
from reportlab.pdfgen import canvas
|
||||
from app.crud import get_shipments, get_shipment_by_id # Import CRUD functions for shipment
|
||||
from app.crud import (
|
||||
get_shipments,
|
||||
get_shipment_by_id,
|
||||
) # Import CRUD functions for shipment
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
def generate_unique_id(db: Session, length: int = 16) -> str:
|
||||
while True:
|
||||
base_string = f"{time.time()}{random.randint(0, 10 ** 6)}"
|
||||
hash_object = hashlib.sha256(base_string.encode())
|
||||
hash_digest = hash_object.hexdigest()
|
||||
unique_id = ''.join(random.choices(hash_digest, k=length))
|
||||
existing_dewar = db.query(DewarModel).filter(DewarModel.unique_id == unique_id).first()
|
||||
unique_id = "".join(random.choices(hash_digest, k=length))
|
||||
existing_dewar = (
|
||||
db.query(DewarModel).filter(DewarModel.unique_id == unique_id).first()
|
||||
)
|
||||
if not existing_dewar:
|
||||
break
|
||||
return unique_id
|
||||
|
||||
|
||||
@router.post("/", response_model=DewarSchema, status_code=status.HTTP_201_CREATED)
|
||||
async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> DewarSchema:
|
||||
async def create_dewar(
|
||||
dewar: DewarCreate, db: Session = Depends(get_db)
|
||||
) -> DewarSchema:
|
||||
try:
|
||||
db_dewar = DewarModel(
|
||||
dewar_name=dewar.dewar_name,
|
||||
@ -96,6 +105,7 @@ async def create_dewar(dewar: DewarCreate, db: Session = Depends(get_db)) -> Dew
|
||||
logging.error(f"Validation error occurred: {e}")
|
||||
raise HTTPException(status_code=400, detail="Validation error")
|
||||
|
||||
|
||||
@router.post("/{dewar_id}/generate-qrcode")
|
||||
async def generate_dewar_qrcode(dewar_id: int, db: Session = Depends(get_db)):
|
||||
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
|
||||
@ -109,7 +119,7 @@ async def generate_dewar_qrcode(dewar_id: int, db: Session = Depends(get_db)):
|
||||
qr = qrcode.QRCode(version=1, box_size=10, border=5)
|
||||
qr.add_data(dewar.unique_id)
|
||||
qr.make(fit=True)
|
||||
img = qr.make_image(fill='black', back_color='white')
|
||||
img = qr.make_image(fill="black", back_color="white")
|
||||
|
||||
buf = io.BytesIO()
|
||||
img.save(buf)
|
||||
@ -120,6 +130,7 @@ async def generate_dewar_qrcode(dewar_id: int, db: Session = Depends(get_db)):
|
||||
|
||||
return {"message": "QR Code generated", "qrcode": dewar.unique_id}
|
||||
|
||||
|
||||
def generate_label(dewar):
|
||||
buffer = BytesIO()
|
||||
# Set page orientation to landscape
|
||||
@ -138,25 +149,36 @@ def generate_label(dewar):
|
||||
|
||||
# Desired logo width in the PDF (you can adjust this size)
|
||||
desired_logo_width = 4 * cm
|
||||
desired_logo_height = desired_logo_width / logo_aspect_ratio # maintain aspect ratio
|
||||
desired_logo_height = (
|
||||
desired_logo_width / logo_aspect_ratio
|
||||
) # maintain aspect ratio
|
||||
|
||||
# Draw header text
|
||||
c.setFont("Helvetica-Bold", 16)
|
||||
c.drawString(2 * cm, page_height - 2 * cm, "Paul Scherrer Institut")
|
||||
|
||||
# Draw the Heidi logo with preserved aspect ratio
|
||||
c.drawImage(png_logo_path, page_width - desired_logo_width - 2 * cm,
|
||||
page_height - desired_logo_height - 2 * cm,
|
||||
width=desired_logo_width, height=desired_logo_height, mask='auto')
|
||||
c.drawImage(
|
||||
png_logo_path,
|
||||
page_width - desired_logo_width - 2 * cm,
|
||||
page_height - desired_logo_height - 2 * cm,
|
||||
width=desired_logo_width,
|
||||
height=desired_logo_height,
|
||||
mask="auto",
|
||||
)
|
||||
|
||||
# Draw details section
|
||||
c.setFont("Helvetica", 12)
|
||||
|
||||
y_position = page_height - 4 * cm # Adjusted to ensure text doesn't overlap with the logo
|
||||
y_position = (
|
||||
page_height - 4 * cm
|
||||
) # Adjusted to ensure text doesn't overlap with the logo
|
||||
line_height = 0.8 * cm
|
||||
|
||||
if dewar.shipment:
|
||||
c.drawString(2 * cm, y_position, f"Shipment Name: {dewar.shipment.shipment_name}")
|
||||
c.drawString(
|
||||
2 * cm, y_position, f"Shipment Name: {dewar.shipment.shipment_name}"
|
||||
)
|
||||
y_position -= line_height
|
||||
|
||||
c.drawString(2 * cm, y_position, f"Dewar Name: {dewar.dewar_name}")
|
||||
@ -167,7 +189,11 @@ def generate_label(dewar):
|
||||
|
||||
if dewar.contact_person:
|
||||
contact_person = dewar.contact_person
|
||||
c.drawString(2 * cm, y_position, f"Contact: {contact_person.firstname} {contact_person.lastname}")
|
||||
c.drawString(
|
||||
2 * cm,
|
||||
y_position,
|
||||
f"Contact: {contact_person.firstname} {contact_person.lastname}",
|
||||
)
|
||||
y_position -= line_height
|
||||
c.drawString(2 * cm, y_position, f"Email: {contact_person.email}")
|
||||
y_position -= line_height
|
||||
@ -191,15 +217,17 @@ def generate_label(dewar):
|
||||
qr = qrcode.QRCode(version=1, box_size=10, border=4)
|
||||
qr.add_data(dewar.unique_id)
|
||||
qr.make(fit=True)
|
||||
qr_img = qr.make_image(fill='black', back_color='white').convert("RGBA")
|
||||
qr_img = qr.make_image(fill="black", back_color="white").convert("RGBA")
|
||||
|
||||
# Save this QR code to a temporary file
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix='.png') as temp_file:
|
||||
qr_img.save(temp_file, format='PNG')
|
||||
with tempfile.NamedTemporaryFile(delete=False, suffix=".png") as temp_file:
|
||||
qr_img.save(temp_file, format="PNG")
|
||||
temp_file_path = temp_file.name
|
||||
|
||||
# Add QR code to PDF
|
||||
c.drawImage(temp_file_path, page_width - 6 * cm, 5 * cm, width=4 * cm, height=4 * cm)
|
||||
c.drawImage(
|
||||
temp_file_path, page_width - 6 * cm, 5 * cm, width=4 * cm, height=4 * cm
|
||||
)
|
||||
|
||||
# Add footer text
|
||||
c.setFont("Helvetica", 10)
|
||||
@ -207,7 +235,9 @@ def generate_label(dewar):
|
||||
|
||||
# Draw border
|
||||
c.setLineWidth(1)
|
||||
c.rect(1 * cm, 1 * cm, page_width - 2 * cm, page_height - 2 * cm) # Adjusted dimensions
|
||||
c.rect(
|
||||
1 * cm, 1 * cm, page_width - 2 * cm, page_height - 2 * cm
|
||||
) # Adjusted dimensions
|
||||
|
||||
# Finalize the canvas
|
||||
c.showPage()
|
||||
@ -220,25 +250,38 @@ def generate_label(dewar):
|
||||
|
||||
return buffer
|
||||
|
||||
|
||||
@router.get("/{dewar_id}/download-label", response_class=Response)
|
||||
async def download_dewar_label(dewar_id: int, db: Session = Depends(get_db)):
|
||||
dewar = db.query(DewarModel).options(
|
||||
joinedload(DewarModel.pucks).joinedload(PuckModel.samples),
|
||||
joinedload(DewarModel.contact_person),
|
||||
joinedload(DewarModel.return_address),
|
||||
joinedload(DewarModel.shipment)
|
||||
).filter(DewarModel.id == dewar_id).first()
|
||||
dewar = (
|
||||
db.query(DewarModel)
|
||||
.options(
|
||||
joinedload(DewarModel.pucks).joinedload(PuckModel.samples),
|
||||
joinedload(DewarModel.contact_person),
|
||||
joinedload(DewarModel.return_address),
|
||||
joinedload(DewarModel.shipment),
|
||||
)
|
||||
.filter(DewarModel.id == dewar_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not dewar:
|
||||
raise HTTPException(status_code=404, detail="Dewar not found")
|
||||
if not dewar.unique_id:
|
||||
raise HTTPException(status_code=404, detail="QR Code not generated for this dewar")
|
||||
raise HTTPException(
|
||||
status_code=404, detail="QR Code not generated for this dewar"
|
||||
)
|
||||
|
||||
buffer = generate_label(dewar)
|
||||
|
||||
return Response(buffer.getvalue(), media_type="application/pdf", headers={
|
||||
"Content-Disposition": f"attachment; filename=dewar_label_{dewar.id}.pdf"
|
||||
})
|
||||
return Response(
|
||||
buffer.getvalue(),
|
||||
media_type="application/pdf",
|
||||
headers={
|
||||
"Content-Disposition": f"attachment; filename=dewar_label_{dewar.id}.pdf"
|
||||
},
|
||||
)
|
||||
|
||||
|
||||
@router.get("/", response_model=List[DewarSchema])
|
||||
async def get_dewars(db: Session = Depends(get_db)):
|
||||
@ -249,13 +292,23 @@ async def get_dewars(db: Session = Depends(get_db)):
|
||||
logging.error(f"Database error occurred: {e}")
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.get("/dewar-types", response_model=List[DewarTypeSchema])
|
||||
def get_dewar_types(db: Session = Depends(get_db)):
|
||||
return db.query(DewarTypeModel).all()
|
||||
|
||||
@router.get("/dewar-types/{type_id}/serial-numbers", response_model=List[DewarSerialNumberSchema])
|
||||
|
||||
@router.get(
|
||||
"/dewar-types/{type_id}/serial-numbers",
|
||||
response_model=List[DewarSerialNumberSchema],
|
||||
)
|
||||
def get_serial_numbers(type_id: int, db: Session = Depends(get_db)):
|
||||
return db.query(DewarSerialNumberModel).filter(DewarSerialNumberModel.dewar_type_id == type_id).all()
|
||||
return (
|
||||
db.query(DewarSerialNumberModel)
|
||||
.filter(DewarSerialNumberModel.dewar_type_id == type_id)
|
||||
.all()
|
||||
)
|
||||
|
||||
|
||||
@router.post("/dewar-types", response_model=DewarTypeSchema)
|
||||
def create_dewar_type(dewar_type: DewarTypeCreate, db: Session = Depends(get_db)):
|
||||
@ -265,14 +318,18 @@ def create_dewar_type(dewar_type: DewarTypeCreate, db: Session = Depends(get_db)
|
||||
db.refresh(db_type)
|
||||
return db_type
|
||||
|
||||
|
||||
@router.post("/dewar-serial-numbers", response_model=DewarSerialNumberSchema)
|
||||
def create_dewar_serial_number(serial_number: DewarSerialNumberCreate, db: Session = Depends(get_db)):
|
||||
def create_dewar_serial_number(
|
||||
serial_number: DewarSerialNumberCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
db_serial = DewarSerialNumberModel(**serial_number.dict())
|
||||
db.add(db_serial)
|
||||
db.commit()
|
||||
db.refresh(db_serial)
|
||||
return db_serial
|
||||
|
||||
|
||||
@router.get("/dewar-serial-numbers", response_model=List[DewarSerialNumberSchema])
|
||||
def get_all_serial_numbers(db: Session = Depends(get_db)):
|
||||
try:
|
||||
@ -282,22 +339,31 @@ def get_all_serial_numbers(db: Session = Depends(get_db)):
|
||||
logging.error(f"Database error occurred: {e}")
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.get("/{dewar_id}", response_model=DewarSchema)
|
||||
async def get_dewar(dewar_id: int, db: Session = Depends(get_db)):
|
||||
dewar = db.query(DewarModel).options(
|
||||
joinedload(DewarModel.pucks).joinedload(PuckModel.samples),
|
||||
joinedload(DewarModel.contact_person),
|
||||
joinedload(DewarModel.return_address),
|
||||
joinedload(DewarModel.shipment)
|
||||
).filter(DewarModel.id == dewar_id).first()
|
||||
dewar = (
|
||||
db.query(DewarModel)
|
||||
.options(
|
||||
joinedload(DewarModel.pucks).joinedload(PuckModel.samples),
|
||||
joinedload(DewarModel.contact_person),
|
||||
joinedload(DewarModel.return_address),
|
||||
joinedload(DewarModel.shipment),
|
||||
)
|
||||
.filter(DewarModel.id == dewar_id)
|
||||
.first()
|
||||
)
|
||||
|
||||
if not dewar:
|
||||
raise HTTPException(status_code=404, detail="Dewar not found")
|
||||
|
||||
return DewarSchema.from_orm(dewar)
|
||||
|
||||
|
||||
@router.put("/{dewar_id}", response_model=DewarSchema)
|
||||
async def update_dewar(dewar_id: int, dewar_update: DewarUpdate, db: Session = Depends(get_db)) -> DewarSchema:
|
||||
async def update_dewar(
|
||||
dewar_id: int, dewar_update: DewarUpdate, db: Session = Depends(get_db)
|
||||
) -> DewarSchema:
|
||||
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
|
||||
|
||||
if not dewar:
|
||||
@ -311,6 +377,7 @@ async def update_dewar(dewar_id: int, dewar_update: DewarUpdate, db: Session = D
|
||||
db.refresh(dewar)
|
||||
return dewar
|
||||
|
||||
|
||||
@router.delete("/{dewar_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_dewar(dewar_id: int, db: Session = Depends(get_db)):
|
||||
dewar = db.query(DewarModel).filter(DewarModel.id == dewar_id).first()
|
||||
@ -322,6 +389,7 @@ async def delete_dewar(dewar_id: int, db: Session = Depends(get_db)):
|
||||
db.commit()
|
||||
return
|
||||
|
||||
|
||||
# New routes for shipments
|
||||
@router.get("/shipments", response_model=List[ShipmentSchema])
|
||||
async def get_all_shipments(db: Session = Depends(get_db)):
|
||||
@ -332,6 +400,7 @@ async def get_all_shipments(db: Session = Depends(get_db)):
|
||||
logging.error(f"Database error occurred: {e}")
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
||||
|
||||
@router.get("/shipments/{id}", response_model=ShipmentSchema)
|
||||
async def get_single_shipment(id: int, db: Session = Depends(get_db)):
|
||||
try:
|
||||
@ -341,4 +410,4 @@ async def get_single_shipment(id: int, db: Session = Depends(get_db)):
|
||||
return shipment
|
||||
except SQLAlchemyError as e:
|
||||
logging.error(f"Database error occurred: {e}")
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
raise HTTPException(status_code=500, detail="Internal server error")
|
||||
|
Reference in New Issue
Block a user