Model for a SessionPool object.
| 39 | |
| 40 | |
| 41 | class SessionPoolModel(BaseModel): |
| 42 | """Model for a SessionPool object.""" |
| 43 | |
| 44 | model_config = ConfigDict(validate_by_name=True, validate_by_alias=True) |
| 45 | |
| 46 | max_pool_size: Annotated[int, Field(alias='maxPoolSize')] |
| 47 | |
| 48 | sessions: Annotated[ |
| 49 | dict[ |
| 50 | str, |
| 51 | Annotated[ |
| 52 | Session, GetPydanticSchema(lambda _, handler: handler(Any)) |
| 53 | ], # handler(Any) is fine - we validate manually in the BeforeValidator |
| 54 | ], |
| 55 | Field(alias='sessions'), |
| 56 | PlainSerializer( |
| 57 | lambda value: [session.get_state().model_dump(by_alias=True) for session in value.values()], |
| 58 | return_type=list, |
| 59 | ), |
| 60 | BeforeValidator( |
| 61 | lambda value: { |
| 62 | session.id: session |
| 63 | for item in value |
| 64 | if (session := Session.from_model(SessionModel.model_validate(item, by_alias=True))) |
| 65 | } |
| 66 | ), |
| 67 | ] |
| 68 | |
| 69 | @computed_field(alias='sessionCount') |
| 70 | @property |
| 71 | def session_count(self) -> int: |
| 72 | """Get the total number of sessions currently maintained in the pool.""" |
| 73 | return len(self.sessions) |
| 74 | |
| 75 | @computed_field(alias='usableSessionCount') |
| 76 | @property |
| 77 | def usable_session_count(self) -> int: |
| 78 | """Get the number of sessions that are currently usable.""" |
| 79 | return len([session for _, session in self.sessions.items() if session.is_usable]) |
| 80 | |
| 81 | @computed_field(alias='retiredSessionCount') |
| 82 | @property |
| 83 | def retired_session_count(self) -> int: |
| 84 | """Get the number of sessions that are no longer usable.""" |
| 85 | return self.session_count - self.usable_session_count |