295 lines
10 KiB
Python
295 lines
10 KiB
Python
from fastapi import FastAPI, HTTPException, Depends, WebSocket
|
|
from fastapi.security import OAuth2PasswordBearer
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
|
from typing import List, Annotated, Optional, Union
|
|
from sustApp.models.models import Base, Members, Payments
|
|
from sustApp.models.database import engine, SessionLocal
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import func
|
|
from datetime import datetime
|
|
from fastapi.encoders import jsonable_encoder
|
|
# from create_payment_report import create_payment_report
|
|
|
|
import asyncio
|
|
import logging
|
|
|
|
|
|
#http://127.0.0.1:8000/docs -> swagger like ui
|
|
#ngrok http http://localhost:8000
|
|
|
|
app = FastAPI()
|
|
|
|
@app.get("/")
|
|
|
|
def read_root():
|
|
return {"Hello": "World"}
|
|
|
|
#oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
|
|
Base.metadata.create_all(bind=engine)
|
|
|
|
origins = [
|
|
"http://localhost.tiangolo.com",
|
|
"https://localhost.tiangolo.com",
|
|
"http://localhost",
|
|
"http://localhost:8080",
|
|
"http://localhost:3000",
|
|
"http://localhost:5173",
|
|
"http://localhost:5174",
|
|
]
|
|
|
|
app.add_middleware(
|
|
CORSMiddleware,
|
|
allow_origins=origins,
|
|
allow_credentials=True,
|
|
allow_methods=["*"],
|
|
allow_headers=["*"],
|
|
)
|
|
|
|
# Store active WebSocket connections
|
|
connected_clients = []
|
|
|
|
# Custom logging handler to send logs via WebSocket
|
|
class WebSocketHandler(logging.Handler):
|
|
def emit(self, record):
|
|
log_message = self.format(record)
|
|
# Send to all connected clients
|
|
asyncio.create_task(broadcast_log(log_message))
|
|
|
|
# Set up logging
|
|
logger = logging.getLogger("app")
|
|
logger.setLevel(logging.INFO)
|
|
ws_handler = WebSocketHandler()
|
|
logger.addHandler(ws_handler)
|
|
|
|
@app.websocket("/ws/logs")
|
|
async def websocket_endpoint(websocket: WebSocket):
|
|
await websocket.accept()
|
|
connected_clients.append(websocket)
|
|
try:
|
|
while True:
|
|
await websocket.receive_text()
|
|
except:
|
|
connected_clients.remove(websocket)
|
|
|
|
async def broadcast_log(message):
|
|
for client in connected_clients[:]: # Copy list to avoid modification during iteration
|
|
try:
|
|
await client.send_json({
|
|
"type": "log",
|
|
"message": message,
|
|
"timestamp": asyncio.get_event_loop().time()
|
|
})
|
|
except:
|
|
connected_clients.remove(client)
|
|
|
|
@app.get("/test")
|
|
async def test_endpoint():
|
|
logger.info("This is a test log message from backend!")
|
|
return {"message": "Check your browser console"}
|
|
|
|
class User(BaseModel):
|
|
username: str
|
|
email: str | None = None
|
|
full_name: str | None = None
|
|
disabled: bool | None = None
|
|
|
|
class DeliveryBase(BaseModel):
|
|
delivery_date: datetime
|
|
delivered_by_id: int | None = None
|
|
|
|
|
|
class PaymentBase(BaseModel):
|
|
payment_date: datetime | None = None
|
|
payment_year: datetime | None = None
|
|
|
|
class MemberBase(BaseModel):
|
|
member_count: int
|
|
member_firstname: str
|
|
member_firstname_spouse: str | None = None
|
|
member_surname: str
|
|
member_suffix: str | None = None
|
|
member_street: str
|
|
member_streetnumber: str | None = None
|
|
member_email: str | None = None
|
|
member_zip: int | None = Field(default=None, description="Member ZIP code")
|
|
#member_zip: Optional[int] = Field(default=None, description="Member ZIP code")
|
|
member_printed: bool | None = None
|
|
member_city: str
|
|
member_active: bool | None = None
|
|
member_gender: str | None = None
|
|
member_connected: int | None = None
|
|
#member_coordinates:
|
|
#payments: Optional[List[PaymentBase]]
|
|
member_deliveries: Optional[List[DeliveryBase]] = []
|
|
member_payments: Optional[List[PaymentBase]] = []
|
|
current_state_code: str | None = None
|
|
current_state_name: str | None = None
|
|
|
|
@model_validator(mode='before')
|
|
@classmethod
|
|
def handle_empty_strings(cls, data):
|
|
"""Convert empty strings to None for int fields"""
|
|
if isinstance(data, dict):
|
|
if data.get('member_zip') == '':
|
|
data['member_zip'] = None
|
|
return data
|
|
|
|
model_config = ConfigDict(from_attributes=True) # Enables SQLAlchemy object conversion
|
|
|
|
def get_db():
|
|
db = SessionLocal()
|
|
try:
|
|
yield db
|
|
finally:
|
|
db.close()
|
|
|
|
db_dependency = Annotated[Session, Depends(get_db)]
|
|
|
|
@app.get("/members/", response_model=List[MemberBase])
|
|
async def read_members(db: db_dependency, skip: int = 0, limit: int = 100):
|
|
members = db.query(Members).offset(skip).limit(limit).all()
|
|
if not members:
|
|
raise HTTPException(status_code=404, detail='No members found')
|
|
|
|
return [
|
|
{
|
|
"id": m.id,
|
|
"member_firstname": m.member_firstname,
|
|
"member_surname": m.member_surname,
|
|
"member_count": m.member_count,
|
|
"member_street": m.member_street,
|
|
"member_zip": m.member_zip,
|
|
"member_city": m.member_city,
|
|
"member_active": m.member_active,
|
|
"member_printed": m.member_printed,
|
|
"member_deliveries": [
|
|
{
|
|
"id": d.id,
|
|
"delivery_date": d.delivery_date, # Correct field
|
|
"delivered_by_id": d.delivered_by_id,
|
|
# Add any other required delivery fields here
|
|
}
|
|
for d in m.member_deliveries
|
|
] if m.member_deliveries else [],
|
|
"member_payments": [
|
|
{
|
|
"id": p.id,
|
|
"payment_date": p.payment_date,
|
|
"payment_year": p.payment_year,
|
|
# Add any other required delivery fields here
|
|
}
|
|
for p in m.member_payments
|
|
] if m.member_payments else []
|
|
}
|
|
for m in members
|
|
]
|
|
|
|
""" @app.get("/members/", response_model=List[MemberBase])
|
|
async def read_members(db: db_dependency, skip: int = 0, limit: int = 100):
|
|
members = db.query(Members).offset(skip).limit(limit).all()
|
|
if not members:
|
|
raise HTTPException(status_code=404, detail='No members found') # Use 404 for not found
|
|
|
|
# DEBUG: Print first member attributes
|
|
if members:
|
|
print("First member:", vars(members[0]))
|
|
print("First member type:", type(members[0]))
|
|
|
|
# TEMP: Return dicts to bypass Pydantic entirely
|
|
#return [{"id": m.id, "name": getattr(m, 'name', None)} for m in members] #
|
|
|
|
# Safe conversion - handles any SQLAlchemy objects
|
|
return jsonable_encoder(members)
|
|
|
|
# return members # Now works with ORM mode enabled """
|
|
|
|
""" @app.get("/members/", response_model=List[MemberBase])
|
|
async def read_members(db: db_dependency, skip: int = 0, limit: int = 100 ):
|
|
members = db.query(Members).offset(skip).limit(limit).all()
|
|
if not members:
|
|
raise HTTPException(status_code=400, detail='No members')
|
|
return members """
|
|
|
|
@app.get("/activememberscount/", response_model=int)
|
|
async def read_activememberscount(db: db_dependency):
|
|
count = db.query(func.sum(Members.member_count)).filter(Members.current_state_code == 'A', Members.member_primary == True).scalar()
|
|
return count or 0
|
|
|
|
@app.get("/inactivememberscount/", response_model=int)
|
|
async def read_inactivememberscount(db: db_dependency):
|
|
count = db.query(func.sum(Members.member_count)).filter(Members.current_state_code == 'I', Members.member_primary == True).scalar()
|
|
return count or 0
|
|
|
|
@app.get("/deactivatedmemberscount/", response_model=int)
|
|
async def read_deactivatedmemberscount(db: db_dependency):
|
|
count = db.query(func.sum(Members.member_count)).filter(Members.current_state_code == 'D', Members.member_primary == True).scalar()
|
|
return count or 0
|
|
|
|
@app.get("/member/{member_id}")
|
|
async def read_member(member_id: int, db: db_dependency): #, token: Annotated[str, Depends(oauth2_scheme)]):
|
|
result = db.query(Members).join(Members.current_state).filter(Members.id == member_id).first()
|
|
if not result:
|
|
raise HTTPException(status_code=400, detail='Member does not exist')
|
|
print(f"Member: {result.member_surname}")
|
|
print(f"State: {result.current_state.state_name}")
|
|
print(f"Description: {result.current_state.description}")
|
|
|
|
return result
|
|
|
|
@app.get("/payments/")
|
|
async def read_all_payments(db: db_dependency):
|
|
result = db.query(Payments).all()
|
|
if not result:
|
|
raise HTTPException(status_code=400, detail=f'No payments found')
|
|
return result
|
|
|
|
@app.get("/payments/{member_id}")
|
|
async def read_payments(member_id: int, db: db_dependency):
|
|
result = db.query(Payments).filter(Payments.member_id == member_id).all()
|
|
if not result:
|
|
raise HTTPException(status_code=400, detail=f'No payments found for member {member_id}')
|
|
return result
|
|
|
|
@app.post("/members/")
|
|
async def create_member(member: MemberBase, db: db_dependency):
|
|
#db_member = models.Members(**member.model_dump())
|
|
db_member = Members(member_firstname=member.member_firstname,
|
|
member_surname=member.member_surname,
|
|
member_active=member.member_active)
|
|
db.add(db_member)
|
|
db.commit()
|
|
|
|
@app.post("/fullmembers/")
|
|
async def create_members(members: List[MemberBase], db: db_dependency):
|
|
for member in members:
|
|
db_member = Members(member_count=member.member_count,
|
|
member_firstname=member.member_firstname,
|
|
member_firstname_spouse=member.member_firstname_spouse,
|
|
member_surname=member.member_surname,
|
|
member_suffix=member.member_suffix,
|
|
member_street=member.member_street,
|
|
member_streetnumber=member.member_streetnumber,
|
|
member_email=member.member_email,
|
|
member_zip=member.member_zip,
|
|
member_city=member.member_city,
|
|
member_active=member.member_active,
|
|
member_gender=member.member_gender,
|
|
member_connected=member.member_connected)
|
|
db.add(db_member)
|
|
#db.commit()
|
|
#db.refresh(db_member)
|
|
if not member.payments is None:
|
|
for payment in member.payments:
|
|
db_payment = Payments(payment_date=payment.payment_date,
|
|
member_id=db_member.id)
|
|
db.add(db_payment)
|
|
db.commit()
|
|
|
|
# @app.get("/api/reports/payments")
|
|
# async def get_payment_report():
|
|
# try:
|
|
# report_path = create_payment_report()
|
|
# return {"message": "Report generated", "path": report_path}
|
|
# except Exception as e:
|
|
# raise HTTPException(status_code=500, detail=str(e)) |