agentgit.database.repositories package

Submodules

agentgit.database.repositories.checkpoint_repository module

Repository for checkpoint database operations.

Handles CRUD operations for checkpoints in the LangGraph rollback agent system.

class agentgit.database.repositories.checkpoint_repository.CheckpointRepository(db_path=None)[source]

Bases: object

Repository for Checkpoint CRUD operations with SQLite.

Manages checkpoints which capture complete agent state at specific points, allowing rollback functionality.

db_path: Path to the SQLite database file.

Example

>>> repo = CheckpointRepository()
>>> checkpoint = Checkpoint(internal_session_id=1, checkpoint_name="Before action")
>>> saved = repo.create(checkpoint)
>>> checkpoints = repo.get_by_internal_session(1)

count_checkpoints(internal_session_id)[source]

Count checkpoints for an internal session.

Parameters:: internal_session_id (int) – The ID of the internal session.
Returns:: {‘total’: n, ‘auto’: n, ‘manual’: n}
Return type:: Dictionary with counts

create(checkpoint)[source]

Create a new checkpoint.

Parameters:: checkpoint (Checkpoint) – Checkpoint object to create.
Return type:: Checkpoint
Returns:: The created checkpoint with id populated.

delete(checkpoint_id)[source]

Delete a checkpoint.

Parameters:: checkpoint_id (int) – The ID of the checkpoint to delete.
Return type:: bool
Returns:: True if deletion successful, False otherwise.

delete_auto_checkpoints(internal_session_id, keep_latest=5)[source]

Delete old automatic checkpoints, keeping only the most recent ones.

Parameters:

internal_session_id (int) – The ID of the internal session.
keep_latest (int) – Number of latest auto checkpoints to keep.

Return type:

int

Returns:

Number of checkpoints deleted.

get_by_id(checkpoint_id)[source]

Get a checkpoint by ID.

Parameters:: checkpoint_id (int) – The ID of the checkpoint to retrieve.
Return type:: Optional[Checkpoint]
Returns:: Checkpoint if found, None otherwise.

get_by_internal_session(internal_session_id, auto_only=False)[source]

Get all checkpoints for an internal session.

Parameters:

internal_session_id (int) – The ID of the internal session.
auto_only (bool) – If True, only return automatic checkpoints.

Return type:

List[Checkpoint]

Returns:

List of Checkpoint objects, ordered by created_at descending.

get_by_user(user_id, limit=None)[source]

Get all checkpoints for a specific user.

Parameters:

user_id (int) – The ID of the user.
limit (Optional[int]) – Optional limit on number of checkpoints to return.

Return type:

List[Checkpoint]

Returns:

List of Checkpoint objects, ordered by created_at descending.

get_checkpoints_with_tools(internal_session_id)[source]

Get checkpoints that have tool invocations.

Parameters:: internal_session_id (int) – The ID of the internal session.
Return type:: List[Checkpoint]
Returns:: List of Checkpoint objects that have tool invocations.

get_latest_checkpoint(internal_session_id)[source]

Get the most recent checkpoint for an internal session.

Parameters:: internal_session_id (int) – The ID of the internal session.
Return type:: Optional[Checkpoint]
Returns:: The latest Checkpoint if found, None otherwise.

search_checkpoints(internal_session_id, search_term)[source]

Search checkpoints by name or content.

Parameters:

internal_session_id (int) – The ID of the internal session.
search_term (str) – Term to search for in checkpoint names.

Return type:

List[Checkpoint]

Returns:

List of matching Checkpoint objects.

update_checkpoint_metadata(checkpoint_id, metadata)[source]

Update the metadata of a checkpoint.

Useful for updating tool track positions or other metadata after creation.

Parameters:

checkpoint_id (int) – The ID of the checkpoint to update.
metadata (Dict) – New metadata to merge with existing metadata.

Return type:

bool

Returns:

True if update successful, False otherwise.

agentgit.database.repositories.external_session_repository module

Repository for external session database operations.

Handles CRUD operations for external sessions in the LangGraph rollback agent system.

class agentgit.database.repositories.external_session_repository.ExternalSessionRepository(db_path=None)[source]

Bases: object

Repository for ExternalSession CRUD operations with SQLite.

Manages external sessions which are the user-visible conversation containers. Each external session can contain multiple internal sessions with branching support.

db_path: Path to the SQLite database file.

Example

>>> repo = ExternalSessionRepository()
>>> session = ExternalSession(user_id=1, session_name="My Chat")
>>> saved_session = repo.create(session)
>>> sessions = repo.get_user_sessions(user_id=1)

add_internal_session(external_session_id, langgraph_session_id)[source]

Add an internal langgraph session to an external session.

Parameters:

external_session_id (int) – The ID of the external session.
langgraph_session_id (str) – The langgraph session ID to add.

Return type:

bool

Returns:

True if successful, False if external session not found.

check_ownership(session_id, user_id)[source]

Check if a user owns a specific session.

Parameters:

session_id (int) – The ID of the session.
user_id (int) – The ID of the user.

Return type:

bool

Returns:

True if the user owns the session, False otherwise.

count_user_sessions(user_id, active_only=False)[source]

Count the number of sessions a user has.

Parameters:

user_id (int) – The ID of the user.
active_only (bool) – If True, only count active sessions.

Return type:

int

Returns:

The number of sessions.

create(session)[source]

Create a new external session.

Parameters:: session (ExternalSession) – ExternalSession object to create.
Return type:: ExternalSession
Returns:: The created session with id populated.
Raises:: sqlite3.IntegrityError – If user_id doesn’t exist.

deactivate(session_id)[source]

Deactivate an external session (soft delete).

Parameters:: session_id (int) – The ID of the session to deactivate.
Return type:: bool
Returns:: True if deactivation successful, False otherwise.

delete(session_id)[source]

Permanently delete an external session.

Parameters:: session_id (int) – The ID of the session to delete.
Return type:: bool
Returns:: True if deletion successful, False otherwise.

Note

This will cascade delete all internal sessions and checkpoints associated with this external session.

get_by_id(session_id)[source]

Get an external session by ID.

Parameters:: session_id (int) – The ID of the session to retrieve.
Return type:: Optional[ExternalSession]
Returns:: ExternalSession if found, None otherwise.

get_by_internal_session(langgraph_session_id)[source]

Find the external session containing a specific internal langgraph session.

Parameters:: langgraph_session_id (str) – The langgraph session ID to search for.
Return type:: Optional[ExternalSession]
Returns:: ExternalSession containing the internal session, None if not found.

get_user_sessions(user_id, active_only=False)[source]

Get all sessions for a user.

Parameters:

user_id (int) – The ID of the user.
active_only (bool) – If True, only return active sessions.

Return type:

List[ExternalSession]

Returns:

List of ExternalSession objects, ordered by created_at descending.

set_current_internal_session(external_session_id, langgraph_session_id)[source]

Set the current internal session for an external session.

Parameters:

external_session_id (int) – The ID of the external session.
langgraph_session_id (str) – The langgraph session ID to set as current.

Return type:

bool

Returns:

True if successful, False if session not found or langgraph_session_id not in list.

update(session)[source]

Update an existing external session.

Updates all session data including internal session IDs and current session.

Parameters:: session (ExternalSession) – ExternalSession object with updated data.
Return type:: bool
Returns:: True if update successful, False if session not found.

agentgit.database.repositories.internal_session_repository module

Repository for internal session database operations.

Handles CRUD operations for internal langgraph sessions in the rollback agent system.

class agentgit.database.repositories.internal_session_repository.InternalSessionRepository(db_path=None)[source]

Bases: object

Repository for InternalSession CRUD operations with SQLite.

Manages internal langgraph sessions which are the actual agent sessions running within external sessions.

db_path: Path to the SQLite database file.

Example

>>> repo = InternalSessionRepository()
>>> session = InternalSession(external_session_id=1, langgraph_session_id="langgraph_123")
>>> saved = repo.create(session)
>>> sessions = repo.get_by_external_session(1)

count_sessions(external_session_id)[source]

Count internal sessions for an external session.

Parameters:: external_session_id (int) – The ID of the external session.
Return type:: int
Returns:: Number of internal sessions.

create(session)[source]

Create a new internal session.

Parameters:: session (InternalSession) – InternalSession object to create.
Return type:: InternalSession
Returns:: The created session with id populated.

delete(session_id)[source]

Delete an internal session.

Parameters:: session_id (int) – The ID of the session to delete.
Return type:: bool
Returns:: True if deletion successful, False otherwise.

get_branch_sessions(parent_session_id)[source]

Get all sessions branched from a parent session.

Parameters:: parent_session_id (int) – The ID of the parent session.
Return type:: List[InternalSession]
Returns:: List of InternalSession objects branched from the parent.

get_by_external_session(external_session_id)[source]

Get all internal sessions for an external session.

Parameters:: external_session_id (int) – The ID of the external session.
Return type:: List[InternalSession]
Returns:: List of InternalSession objects, ordered by created_at descending.

get_by_id(session_id)[source]

Get an internal session by ID.

Parameters:: session_id (int) – The ID of the session to retrieve.
Return type:: Optional[InternalSession]
Returns:: InternalSession if found, None otherwise.

get_by_langgraph_session_id(langgraph_session_id)[source]

Get an internal session by langgraph session ID.

Parameters:: langgraph_session_id (str) – The langgraph session ID.
Return type:: Optional[InternalSession]
Returns:: InternalSession if found, None otherwise.

get_current_session(external_session_id)[source]

Get the current internal session for an external session.

Parameters:: external_session_id (int) – The ID of the external session.
Return type:: Optional[InternalSession]
Returns:: The current InternalSession if found, None otherwise.

get_session_lineage(session_id)[source]

Get the lineage of a session (path from root to this session).

Parameters:: session_id (int) – The ID of the session.
Return type:: List[InternalSession]
Returns:: List of InternalSession objects from root to current session.

set_current_session(session_id)[source]

Set an internal session as the current one for its external session.

Parameters:: session_id (int) – The ID of the session to set as current.
Return type:: bool
Returns:: True if successful, False if session not found.

update(session)[source]

Update an existing internal session.

Updates session state and conversation history.

Parameters:: session (InternalSession) – InternalSession object with updated data.
Return type:: bool
Returns:: True if update successful, False if session not found.

update_tool_count(session_id, increment=1)[source]

Update the tool invocation count for a session.

Parameters:

session_id (int) – The ID of the session.
increment (int) – Amount to increment by.

Return type:

bool

Returns:

True if update successful, False otherwise.

agentgit.database.repositories.user_repository module

User repository for database operations.

Provides ORM functionality for User entities with SQLite backend.

class agentgit.database.repositories.user_repository.UserRepository(db_path=None)[source]

Bases: object

Repository for User CRUD operations with SQLite.

This class handles all database operations for User entities, including automatic initialization of the database schema and creation of the default admin user (rootusr).

db_path: Path to the SQLite database file.

Example

>>> repo = UserRepository()
>>> user = User(username="alice")
>>> user.set_password("secret")
>>> saved_user = repo.save(user)
>>> found_user = repo.find_by_username("alice")
>>> found_user.verify_password("secret")
True

cleanup_inactive_sessions(user_id, active_session_ids)[source]

Clean up inactive sessions for a user.

Removes session IDs that are no longer active from the user’s active_sessions list.

Parameters:

user_id (int) – The ID of the user.
active_session_ids (List[int]) – List of session IDs that are still active.

Return type:

bool

Returns:

True if cleanup was performed, False otherwise.

delete(user_id)[source]

Delete a user from the database.

Parameters:: user_id (int) – The ID of the user to delete.
Return type:: bool
Returns:: True if a user was deleted, False if no user found.

find_all()[source]

Retrieve all users from the database.

Return type:: List[User]
Returns:: List of all User objects in the database.

find_by_api_key(api_key)[source]

Find a user by their API key.

Parameters:: api_key (str) – The API key to search for.
Return type:: Optional[User]
Returns:: User object if found, None otherwise.

find_by_id(user_id)[source]

Find a user by their database ID.

Parameters:: user_id (int) – The unique identifier of the user.
Return type:: Optional[User]
Returns:: User object if found, None otherwise.

find_by_username(username)[source]

Find a user by their username.

Parameters:: username (str) – The username to search for.
Return type:: Optional[User]
Returns:: User object if found, None otherwise.

get_user_sessions(user_id)[source]

Get all active session IDs for a user.

Parameters:: user_id (int) – The ID of the user.
Return type:: List[int]
Returns:: List of active session IDs.

save(user)[source]

Save or update a user in the database.

Performs insert if user.id is None, otherwise updates existing record. Stores both structured fields and full JSON representation for flexibility.

Parameters:: user (User) – User object to save.
Return type:: User
Returns:: The saved User object with id populated if newly created.

Note

Password hash is stored separately from the JSON data for security.

update_api_key(user_id, api_key)[source]

Update or remove a user’s API key.

Parameters:

user_id (int) – The ID of the user to update.
api_key (Optional[str]) – New API key or None to remove.

Return type:

bool

Returns:

True if updated successfully, False otherwise.

update_last_login(user_id)[source]

Update the last login timestamp for a user.

Parameters:: user_id (int) – The ID of the user to update.
Return type:: bool
Returns:: True if updated successfully, False otherwise.

update_user_preferences(user_id, preferences)[source]

Update user preferences.

Parameters:

user_id (int) – The ID of the user.
preferences (dict) – Dictionary of preferences to update.

Return type:

bool

Returns:

True if updated successfully, False otherwise.

update_user_sessions(user_id, session_ids)[source]

Update the active sessions list for a user.

Parameters:

user_id (int) – The ID of the user.
session_ids (List[int]) – New list of active session IDs.

Return type:

bool

Returns:

True if updated successfully, False otherwise.

agentgit.database.repositories package

Submodules

agentgit.database.repositories.checkpoint_repository module

agentgit.database.repositories.external_session_repository module

agentgit.database.repositories.internal_session_repository module

agentgit.database.repositories.user_repository module

Module contents