(
request: Request,
)
| 453 | tags=["openui/session"], |
| 454 | ) |
| 455 | async def get_session( |
| 456 | request: Request, |
| 457 | ): |
| 458 | session_id = request.session.get("session_id") |
| 459 | if session_id is None: |
| 460 | if config.ENV == config.Env.LOCAL: |
| 461 | # Give local users a session automatically |
| 462 | session_id = session_store.generate_session_id() |
| 463 | request.session["session_id"] = session_id |
| 464 | user_id = uuid.uuid4() |
| 465 | try: |
| 466 | user = User.get_or_none(User.username == getpass.getuser()) |
| 467 | if user is None: |
| 468 | user = User.create( |
| 469 | username=getpass.getuser(), |
| 470 | created_at=datetime.now(), |
| 471 | id=user_id.bytes, |
| 472 | ) |
| 473 | else: |
| 474 | user_id = user.id |
| 475 | except IntegrityError: |
| 476 | user = User.get(User.username == getpass.getuser()) |
| 477 | user_id = user.id |
| 478 | if user.email is None: |
| 479 | user.email = get_git_user_email() |
| 480 | user.save() |
| 481 | request.session["user_id"] = str(user_id) |
| 482 | session_store.write( |
| 483 | request.session["session_id"], |
| 484 | str(user_id), |
| 485 | SessionData( |
| 486 | username=user.username, |
| 487 | token_count=0, |
| 488 | max_tokens=config.MAX_TOKENS, |
| 489 | email=user.email, |
| 490 | ), |
| 491 | ) |
| 492 | else: |
| 493 | raise HTTPException(status_code=404, detail="No session found") |
| 494 | session_data = session_store.get(session_id) |
| 495 | return JSONResponse( |
| 496 | content=session_data.model_dump(), |
| 497 | status_code=200, |
| 498 | ) |
| 499 | |
| 500 | |
| 501 | @router.delete( |
nothing calls this directly
no test coverage detected