Persistence Layer (database)¶

This package handles all data storage and retrieval operations, including the ORM models and the repository pattern implementation.

Database¶

kusibot.database.db.init_db(app)[source]¶

Initialising DB (SQLAlchemy) with Flask app.

Parameters:: app – The Flask app instance.

kusibot.database.db.initialise_data(app, db)[source]¶

Initialises the database creating all its tables and the UNIQUE professional user who has access to the non-professional users conversations and insights.

Parameters:

app – The Flask app instance.
db – The SQLAlchemy instance.

Models¶

class kusibot.database.models.Assessment(**kwargs)[source]¶

Assessment model for the database.

id¶

Assessment ID.

Type:: int

user_id¶

User ID that took the assessment.

Type:: int

assessment_type¶

Type of assessment (e.g., ‘PHQ-9’).

Type:: str

message_trigger¶

Trigger message for the assessment.

Type:: str

start_time¶

Start time of the assessment.

Type:: datetime

end_time¶

End time of the assessment.

Type:: datetime

total_score¶

Total score of the assessment.

Type:: int

interpretation¶

Interpretation of the assessment.

Type:: str

questions¶

Relationship with the AssessmentQuestion model.

Type:: Relationship

query: t.ClassVar[Query]¶: A SQLAlchemy query for a model. Equivalent to db.session.query(Model). Can be customized per-model by overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

class kusibot.database.models.AssessmentQuestion(**kwargs)[source]¶

AssessmentQuestion model for the database.

id¶

AssessmentQuestion ID.

Type:: int

assessment_id¶

Assessment ID that the question belongs to.

Type:: int

question_number¶

Question number.

Type:: int

question_text¶

Question text.

Type:: str

user_response¶

User response to the question.

Type:: str

categorized_value¶

Mapped to standard scale.

Type:: int

timestamp¶

Question timestamp

Type:: datetime

query: t.ClassVar[Query]¶: A SQLAlchemy query for a model. Equivalent to db.session.query(Model). Can be customized per-model by overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

class kusibot.database.models.Conversation(**kwargs)[source]¶

Conversation model for the database.

id¶

Conversation ID.

Type:: int

user_id¶

User ID that started the conversation.

Type:: int

created_at¶

Conversation creation date.

Type:: datetime

finished_at¶

Conversation finish date.

Type:: datetime

messages¶

Relationship with the Message model.

Type:: Relationship

query: t.ClassVar[Query]¶: A SQLAlchemy query for a model. Equivalent to db.session.query(Model). Can be customized per-model by overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

class kusibot.database.models.Message(**kwargs)[source]¶

Message model for the database.

id¶

Message ID.

Type:: int

conversation_id¶

Conversation ID that the message belongs to.

Type:: int

text¶

Message text.

Type:: str

timestamp¶

Message timestamp.

Type:: datetime

is_user¶

True if the message is from the user, False if it’s from chatbot.

Type:: bool

intent¶

Intent of the message if detected.

Type:: str

agent_type¶

Type of agent that answered the message (Conversarion or Assesment agent).

Type:: str

query: t.ClassVar[Query]¶: A SQLAlchemy query for a model. Equivalent to db.session.query(Model). Can be customized per-model by overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

class kusibot.database.models.User(**kwargs)[source]¶

User model for the database.

id¶

User ID.

Type:: int

username¶

User’s username.

Type:: str

email¶

User’s email.

Type:: str

password¶

User’s password.

Type:: str

created_at¶

User’s account creation date.

Type:: datetime

is_professional¶

True if the user is a professional (can go into dashboard), False if it’s a simple user.

Type:: bool

conversations¶

Relationship with the Conversation model.

Type:: Relationship

assessments¶

Relationship with the Assessment model.

Type:: Relationship

check_password(password)[source]¶

Check if the provided password matches the user’s password.

Parameters:: password (str) – The password to check.
Returns:: True if the password matches, False otherwise.
Return type:: bool

query: t.ClassVar[Query]¶: A SQLAlchemy query for a model. Equivalent to db.session.query(Model). Can be customized per-model by overriding query_class.

Warning

The query interface is considered legacy in SQLAlchemy. Prefer using session.execute(select()) instead.

Repositories¶

class kusibot.database.db_repositories.AssessmentQuestionRepository[source]¶

Manages all data access logic for the AssessmentQuestion model.

get_question_by_assessment_id(assessment_id)[source]¶

Retrieve all questions for a given assessment, ordered by question number.

Parameters:: assessment_id – The ID of the assessment whose questions are to be retrieved.
Returns:: A list of AssessmentQuestion objects.
Return type:: list

save_assessment_question(assessment_id, question_number, question_text, user_response, categorized_value)[source]¶

Save an assessment question to the database.

Parameters:

assessment_id – The ID of the assessment to which the question belongs.
question_number – The number of the question in the assessment.
question_text – The text of the question.
user_response – The user’s free-text response to the question.
categorized_value – The categorized value of the user’s response (depends on questionnaire).

class kusibot.database.db_repositories.AssessmentRepository[source]¶

Manages all data access logic for the Assessment model.

calculate_total_score(assessment_id)[source]¶

Calculate the total score of an assessment by summing the values of its questions.

Parameters:: assessment_id – The ID of the assessment for which to calculate the total score.
Returns:: The total score of the assessment, or 0 if no questions are found or an error occurred.
Return type:: int

create_assessment(message_trigger, user_id, assessment_type, state)[source]¶

Create a new assessment for a given user.

Parameters:

user_id – The ID of the user for whom the assessment is to be created.
assessment_type – The type of the assessment (e.g., “PHQ9”, “GAD7”).
state – The initial state of the assessment (e.g., “AskingQuestion state”).

Returns:

The newly created assessment object if successful, otherwise None.

Return type:

Assessment

end_assessment(user_id)[source]¶

End the current assessment for a given user by setting its end_time.

Parameters:: user_id – The ID of the user whose assessment is to be ended.

get_assessment(assessment_id)[source]¶

Retrieve an assessment by its ID.

Parameters:: assessment_id – The ID of the assessment to retrieve.
Returns:: The assessment object if found, otherwise None.
Return type:: Assessment

get_assessments_by_user_id(user_id)[source]¶

Retrieve all assessments for a given user, ordered by start time in descending order.

Parameters:: user_id – The ID of the user whose assessments are to be retrieved.
Returns:: A list of Assessment objects.
Return type:: list

get_current_assessment(user_id)[source]¶

Retrieve the current assessment (not finished) for a given user.

Parameters:: user_id – The ID of the user whose current assessment is to be retrieved.
Returns:: The current assessment object if found, otherwise None.
Return type:: Assessment

is_assessment_active(user_id)[source]¶

Check if there is an active assessment for the given user.

Parameters:: user_id – The ID of the user to check.
Returns:: True if an active assessment exists, False otherwise.
Return type:: bool

update_assessment(assessment_id, **kwargs)[source]¶

Update an existing assessment with new values.

Parameters:

assessment_id – The ID of the assessment to update.
**kwargs – The fields to update and their new values.

class kusibot.database.db_repositories.ConversationRepository[source]¶

Manages all data access logic for the Conversation model.

create_conversation(user_id)[source]¶

Create a new conversation for a given user.

Parameters:: user_id – The ID of the user for whom the conversation is to be created.
Returns:: The newly created conversation object if successful, otherwise None.
Return type:: Conversation

end_conversation(conv_id)[source]¶

End the given conversation by setting its finished_at timestamp.

Parameters:: conv_id – The ID of the conversation to end.

get_all_conversations_by_user_id(user_id)[source]¶

Retrieve all conversations for a given user, ordered by creation time in descending order.

Parameters:: user_id – The ID of the user whose conversations are to be retrieved.
Returns:: A list of Conversation objects.
Return type:: list

get_conversation(conv_id)[source]¶

Retrieve a conversation by its ID.

Parameters:: conv_id – The ID of the conversation to retrieve.
Returns:: The conversation object if found, otherwise None.
Return type:: Conversation

get_current_conversation_by_user_id(user_id)[source]¶

Retrieve the current conversation (not finished) for a given user.

Parameters:: user_id – The ID of the user whose current conversation is to be retrieved.
Returns:: The current conversation object if found, otherwise None.
Return type:: Conversation

get_last_conversation_by_user_id(user_id)[source]¶

Retrieve the last conversation (finished or not) for a given user.

Parameters:: user_id – The ID of the user whose last conversation is to be retrieved.
Returns:: The last conversation object if found, otherwise None.
Return type:: Conversation

class kusibot.database.db_repositories.MessageRepository[source]¶

Manages all data access logic for the Message model.

get_limited_messages(conv_id, limit)[source]¶

Retrieve the last <limit> messages from a conversation, ordered by timestamp.

Parameters:

conv_id – The ID of the conversation from which to retrieve messages.
limit – The maximum number of messages to retrieve.

Returns:

A list of Message objects.

Return type:

list

get_messages_by_conversation_id(conv_id)[source]¶

Retrieve all messages from a conversation, ordered by timestamp.

Parameters:: conv_id – The ID of the conversation from which to retrieve messages.
Returns:: A list of Message objects.
Return type:: list

save_chatbot_message(conv_id, msg, intent=None, agent_type='Conversation')[source]¶

Save a chatbot message to the conversation stored in database.

Parameters:

conv_id – The ID of the conversation to which the message belongs.
msg – The text of the chatbot message.
intent – The intent of the message, if applicable (generally, it does not).
agent_type – The type of agent sending the message (default is the ConversationAgent).

save_user_message(conv_id, msg, intent=None)[source]¶

Save a user message to the conversation stored in database.

Parameters:

conv_id – The ID of the conversation to which the message belongs.
msg – The text of the user message.
intent – The intent of the message, if applicable.

class kusibot.database.db_repositories.UserRepository[source]¶

Manages all data access logic for the User model.

add_user(username, email, hashed_password, is_professional)[source]¶

Add a new user to the database.

Parameters:

username (str) – The username of the new user.
email (str) – The email of the new user.
hashed_password (str) – The hashed password of the new user.
is_professional (bool) – Whether the user is a professional user or not.

Returns:

The newly created user object if successful, otherwise None.

Return type:

User

get_non_professional_users()[source]¶

Retrieve all non-professional users from the database.

Returns:: A list of non-professional User objects.
Return type:: list

get_user_by_email(email)[source]¶

Retrieve a user by their email.

Parameters:: email (str) – The email of the user to retrieve.
Returns:: The user object if found, otherwise None.
Return type:: User

get_user_by_username(username)[source]¶

Retrieve a user by their username.

Parameters:: username (str) – The username of the user to retrieve.
Returns:: The user object if found, otherwise None.
Return type:: User