42 lines
1.6 KiB
Python
42 lines
1.6 KiB
Python
"""radacct — accounting data. Read-only: FreeRADIUS owns writes to this table."""
|
|
from fastapi import APIRouter, Depends, Query
|
|
from sqlalchemy import select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from .. import crud
|
|
from ..database import get_db
|
|
from ..models import RadAcct
|
|
from ..pagination import Page, PageParams
|
|
from ..schemas import RadAcctOut
|
|
|
|
router = APIRouter(prefix="/radacct", tags=["radacct (read-only)"])
|
|
|
|
|
|
@router.get("", response_model=Page[RadAcctOut])
|
|
def list_acct(
|
|
page: PageParams = Depends(),
|
|
username: str | None = None,
|
|
nasipaddress: str | None = None,
|
|
active: bool | None = Query(default=None, description="true = sessions with no stop time"),
|
|
db: Session = Depends(get_db),
|
|
):
|
|
filters = {"username": username, "nasipaddress": nasipaddress}
|
|
total = crud.count_rows(db, RadAcct, filters)
|
|
|
|
stmt = select(RadAcct)
|
|
for field, value in filters.items():
|
|
if value is not None:
|
|
stmt = stmt.where(getattr(RadAcct, field) == value)
|
|
if active is True:
|
|
stmt = stmt.where(RadAcct.acctstoptime.is_(None))
|
|
elif active is False:
|
|
stmt = stmt.where(RadAcct.acctstoptime.is_not(None))
|
|
stmt = stmt.order_by(RadAcct.acctstarttime.desc()).limit(page.limit).offset(page.offset)
|
|
rows = list(db.execute(stmt).scalars().all())
|
|
return Page(total=total, limit=page.limit, offset=page.offset, items=rows)
|
|
|
|
|
|
@router.get("/{radacctid}", response_model=RadAcctOut)
|
|
def get_acct(radacctid: int, db: Session = Depends(get_db)):
|
|
return crud.get_row_or_404(db, RadAcct, radacctid, pk_field="radacctid")
|