Services

src.core.services.auth_service.build_claims(user: User) → dict

Builds JWT claims from a user object.

Parameters:

user (User) – User object

Returns:

JWT claims

Return type:

dict

src.core.services.auth_service.check_password(user: User, password: str) → bool

Checks if a password matches the user’s password hash.

Parameters:

• user (User) – User object

• password (str) – Password to check

Returns:

True if password matches, False otherwise

Return type:

bool

src.core.services.auth_service.create_tokens(user: User, access_expires_seconds: int, refresh_expires_seconds: int) → dict

Creates access and refresh tokens for a user.

Parameters:

• user (User) – User object

• access_expires_seconds (int) – Access token expiration time in seconds

• refresh_expires_seconds (int) – Refresh token expiration time in seconds

Returns:

A dictionary containing access and refresh tokens

Return type:

dict

src.core.services.auth_service.profile_from_jwt(identity: str | None, jwt_payload: dict) → dict

Builds a user profile from JWT payload.

Parameters:

• identity (str | None) – User identity

• jwt_payload (dict) – JWT payload

Returns:

User profile

Return type:

dict

src.core.services.auth_service.refresh_access_token(identity: str, claims: dict, access_expires_seconds: int) → dict

Refreshes an access token for a user.

Parameters:

• identity (str) – User identity

• claims (dict) – JWT claims

• access_expires_seconds (int) – Access token expiration time in seconds

Returns:

A dictionary containing a new access token

Return type:

dict

src.core.services.auth_service.revoke_token(jti: str | None) → None

Revokes a JWT token.

Parameters:

jti (str | None) – JWT token ID

src.core.services.auth_service.verify_user_active_role(user: User) → bool

Verifies if a user has an active role.

Parameters:

user (User) – User object

Returns:

True if user has an active role, False otherwise

Return type:

bool

src.core.services.jwt_blocklist.is_token_revoked(jti: str | None) → bool

Checks if a JWT token is revoked.

Parameters:

jti (str | None) – The JWT token identifier to check.

Returns:

True if the token is revoked, False otherwise.

Return type:

bool

src.core.services.jwt_blocklist.revoke_token(jti: str | None) → None

Adds a JWT token to the blocklist.

Parameters:

jti (str | None) – The JWT token identifier to be revoked.

src.core.services.metrics_service.get_dashboard_metrics() → dict[str, Any]

Return aggregated metrics for the admin dashboard.

Returns:

Dictionary with keys: cards, charts.

Return type:

dict[str, Any]

src.core.services.questions_service.create_question(data: dict) → Question

Create question.

Parameters:

data (dict) – Data to create question.

Returns:

Created question.

Return type:

Question

src.core.services.questions_service.get_question(question_id: int) → Question | None

Get question by ID.

Parameters:

question_id (int) – Question ID.

Returns:

Question or None if not found.

Return type:

Question | None

src.core.services.questions_service.list_questions(survey_id: int | None = None) → list[Question]

List questions.

Parameters:

survey_id (int | None, optional) – Survey ID. Defaults to None.

Returns:

List of questions.

Return type:

list[Question]

src.core.services.questions_service.set_question_state(question_id: int, state: bool) → Question | None

Set question state.

Parameters:

• question_id (int) – Question ID.

• state (bool) – State to set.

Returns:

Updated question or None if not found.

Return type:

Question | None

src.core.services.questions_service.update_question(question_id: int, data: dict) → Question | None

Update question.

Parameters:

• question_id (int) – Question ID.

• data (dict) – Data to update question.

Returns:

Updated question or None if not found.

Return type:

Question | None

src.core.services.reports_service.export_summary_csv(*, summary: dict) → str

Export a summary dict to CSV with rows per answer value.

Parameters:

summary (dict) – Formatted summary.

Returns:

CSV content.

Return type:

str

src.core.services.reports_service.get_survey_summary(survey_id: int, *, team_id: int | None = None, date_from: datetime | None = None, date_to: datetime | None = None) → dict

Aggregate responses for a survey, optionally filtered by team and date range.

Parameters:

• survey_id (int) – ID of the survey.

• team_id (int | None, optional) – ID of the team. Defaults to None.

• date_from (datetime | None, optional) – Start of the date range. Defaults to None.

• date_to (datetime | None, optional) – End of the date range. Defaults to None.

Returns:

Formatted summary.

Return type:

dict

src.core.services.reports_service.get_team_summary(team_id: int, *, survey_id: int | None = None, date_from: datetime | None = None, date_to: datetime | None = None) → dict

Aggregate responses for a team, optionally filtered by survey and date range.

Parameters:

• team_id (int) – ID of the team.

• survey_id (int | None, optional) – ID of the survey. Defaults to None.

• date_from (datetime | None, optional) – Start of the date range. Defaults to None.

• date_to (datetime | None, optional) – End of the date range. Defaults to None.

Returns:

Formatted summary.

Return type:

dict

src.core.services.responses_service.get_anonymous_survey(*, token_str: str, survey_id: int | None = None) → dict

Get survey and active questions from a single-use token.

Validates membership, token freshness, and survey status (active and anonymous).

Parameters:

• token_str (str) – Survey token.

• survey_id (int | None) – Optional survey ID to validate membership.

Raises:

• LookupError – If token not found.

• ValueError – If token is not active or not anonymous.

Returns:

{ survey: {id, title, description}, token: { id, token, is_used, expires_at, used_at, survey_id }, questions: [ … ] }

Return type:

dict

src.core.services.responses_service.submit_anonymous_responses(*, token_str: str, responses: Iterable[dict], survey_id: int | None = None) → dict

Submit anonymous responses using a one-time survey token.

Implements idempotence: if the token has already been used or responses already exist associated with the token, rejects double sending.

Parameters:

• token_str (str) – Survey token (UUID/string)

• responses (Iterable[dict]) – List of responses with {question_id, answer}

• survey_id (int | None, optional) – Additional membership validation. Defaults to None.

Returns:

Summary of operation {saved_count, token_id, survey_id}

Return type:

dict

src.core.services.surveys_service.create_survey(data: dict) → Survey

Create survey.

Parameters:

data (dict) – Data to create survey.

Returns:

Created survey.

Return type:

Survey

src.core.services.surveys_service.deactivate_expired_surveys(now: datetime | None = None) → dict

Set state=False for surveys whose expires_at is in the past.

Parameters:

now (datetime | None) – Reference datetime (UTC). Defaults to current UTC.

Returns:

{“matched”: int, “updated”: int}

Return type:

dict

src.core.services.surveys_service.get_survey(survey_id: int) → Survey | None

Get survey by ID.

Parameters:

survey_id (int) – Survey ID.

Returns:

Survey or None if not found.

Return type:

Survey | None

src.core.services.surveys_service.list_surveys(team_id: int | None = None) → list[Survey]

List surveys.

Parameters:

team_id (int | None, optional) – Team ID. Defaults to None.

Returns:

List of surveys.

Return type:

list[Survey]

src.core.services.surveys_service.set_survey_state(survey_id: int, state: bool) → Survey | None

Set survey state.

Parameters:

• survey_id (int) – Survey ID.

• state (bool) – State to set.

Returns:

Updated survey or None if not found.

Return type:

Survey | None

src.core.services.surveys_service.update_survey(survey_id: int, data: dict) → Survey | None

Update survey.

Parameters:

• survey_id (int) – Survey ID.

• data (dict) – Data to update survey.

Returns:

Updated survey or None if not found.

Return type:

Survey | None

src.core.services.teams_service.create_team(data: dict) → Team

Create team.

Parameters:

data (dict) – Data to create team.

Returns:

Created team.

Return type:

Team

src.core.services.teams_service.get_team(team_id: int) → Team | None

Get team by ID.

Parameters:

team_id (int) – Team ID.

Returns:

Team or None if not found.

Return type:

Team | None

src.core.services.teams_service.list_teams() → list[Team]

List teams.

Returns:

List of teams.

Return type:

list[Team]

src.core.services.teams_service.set_team_state(team_id: int, state: bool) → Team | None

Set team state.

Parameters:

• team_id (int) – Team ID.

• state (bool) – State to set.

Returns:

Updated team or None if not found.

Return type:

Team | None

src.core.services.teams_service.update_team(team_id: int, data: dict) → Team | None

Update team.

Parameters:

• team_id (int) – Team ID.

• data (dict) – Data to update.

Returns:

Updated team or None if not found.

Return type:

Team | None

src.core.services.tokens_service.cleanup_expired_tokens(*, survey_id: int | None = None, team_id: int | None = None, dry_run: bool = False, older_than: datetime | None = None) → dict

Clean up expired tokens.

Parameters:

• survey_id (int | None) – Optional filter by survey.

• team_id (int | None) – Optional filter by team.

• dry_run (bool) – If True, do not delete; only return the count.

• older_than (datetime | None) – Threshold time; defaults to now (UTC).

Returns:

{“matched”: int, “deleted”: int}

Return type:

dict

src.core.services.tokens_service.export_tokens_csv(survey_id: int, *, is_used: bool | None = None, include_expired: bool = True) → str

Export survey tokens to CSV.

Parameters:

• survey_id (int) – Survey ID

• is_used (bool | None, optional) – Filter by used status. Defaults to None.

• include_expired (bool, optional) – Include expired tokens. Defaults to True.

Returns:

CSV string

Return type:

str

src.core.services.tokens_service.generate_tokens_for_survey(survey_id: int, count: int = 1, *, expires_at: datetime.datetime(2025, 12, 9, 8, 40, 8, 737575, tzinfo=datetime.timezone.utc), team_id: int | None = None, employee_identifiers: Iterable[str] | None = None) → list[SurveyToken]

Generate one-time tokens for a survey.

• Ensures uniqueness via DB unique constraint with retry on collision.

• Defaults team_id to survey.team_id if not provided.

• Optionally attach employee_identifiers (length must match count if provided).

Parameters:

• survey_id (int) – Survey ID

• expires_at (datetime.now) – Expiration datetime

• count (int, optional) – Number of tokens to generate. Defaults to 1.

• team_id (int | None, optional) – Team ID. Defaults to None.

• employee_identifiers (Iterable[str] | None, optional) – Employee identifiers. Defaults to None.

Raises:

• ValueError – If count is not > 0

• LookupError – If survey_id does not exist

• ValueError – If employee_identifiers length does not match count

• RuntimeError – If failed to generate unique tokens after max retries

Returns:

List of generated SurveyToken instances

Return type:

list[SurveyToken]

src.core.services.tokens_service.is_token_valid_for_submission(token_str: str, survey_id: int | None = None) → bool

Validate token for anonymous submission. Checks not expired and not used. Optionally assures it belongs to a given survey_id.

Parameters:

• token_str (str) – Token string

• survey_id (int | None, optional) – Survey ID. Defaults to None.

Returns:

True if valid for submission, False otherwise

Return type:

bool

src.core.services.tokens_service.list_tokens(survey_id: int, *, is_used: bool | None = None, include_expired: bool = True) → list[dict]

List survey tokens.

Parameters:

• survey_id (int) – Survey ID

• is_used (bool | None, optional) – Filter by used status. Defaults to None.

• include_expired (bool, optional) – Include expired tokens. Defaults to True.

Returns:

List of token dictionaries

Return type:

list[dict]

src.core.services.tokens_service.mark_token_used(token_str: str) → bool

Mark token as used.

Parameters:

token_str (str) – Token string

Returns:

True if marked as used, False if not found or already used

Return type:

bool