Skip to content

Database Operations

loglife.app.db.tables.users

User table operations and data model.

This module defines the User data class and the UsersTable class for handling database interactions related to users.

User dataclass

User data model representing a row in the users table.

Source code in src/loglife/app/db/tables/users.py 
12
13
14
15
16
17
18
19
20
21
22
23
@dataclass
class User:
    """User data model representing a row in the users table."""

    id: int
    phone_number: str
    timezone: str
    created_at: datetime
    send_transcript_file: int  # 0 or 1
    state: str | None
    state_data: str | None
    referred_by_id: int | None

UsersTable

Handles database operations for the users table.

Source code in src/loglife/app/db/tables/users.py 
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class UsersTable:
    """Handles database operations for the users table."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        """Initialize the UsersTable with a database connection."""
        self._conn = conn

    def get(self, user_id: int) -> User | None:
        """Retrieve a user by their ID."""
        query = "SELECT * FROM users WHERE id = ?"
        row = self._conn.execute(query, (user_id,)).fetchone()

        if row:
            return self._row_to_model(row)
        return None

    def get_by_phone(self, phone_number: str) -> User | None:
        """Retrieve a user by their phone number."""
        query = "SELECT * FROM users WHERE phone_number = ?"
        row = self._conn.execute(query, (phone_number,)).fetchone()

        if row:
            return self._row_to_model(row)
        return None

    def get_all(self) -> list[User]:
        """Retrieve all users."""
        query = "SELECT * FROM users"
        rows = self._conn.execute(query).fetchall()
        return [self._row_to_model(row) for row in rows]

    def create(
        self,
        phone_number: str,
        timezone: str,
        referred_by_id: int | None = None,
    ) -> User:
        """Create a new user record."""
        query = "INSERT INTO users (phone_number, timezone, referred_by_id) VALUES (?, ?, ?)"
        cursor = self._conn.execute(query, (phone_number, timezone, referred_by_id))
        # We need to fetch the created user to return the full model with ID and timestamps
        # The result of get() will not be None here because we just inserted it.
        return self.get(cursor.lastrowid)  # type: ignore[arg-type]

    def update(
        self,
        user_id: int,
        phone_number: str | None = None,
        timezone: str | None = None,
        send_transcript_file: int | None = None,
        referred_by_id: int | None = None,
    ) -> User | None:
        """Update a user record with provided fields."""
        updates = []
        params = []

        if phone_number is not None:
            updates.append("phone_number = ?")
            params.append(phone_number)

        if timezone is not None:
            updates.append("timezone = ?")
            params.append(timezone)

        if send_transcript_file is not None:
            updates.append("send_transcript_file = ?")
            params.append(send_transcript_file)

        if referred_by_id is not None:
            updates.append("referred_by_id = ?")
            params.append(referred_by_id)

        if not updates:
            return self.get(user_id)

        params.append(user_id)
        query = f"UPDATE users SET {', '.join(updates)} WHERE id = ?"
        self._conn.execute(query, params)

        return self.get(user_id)

    def set_state(self, user_id: int, state: str | None, state_data: str | None = None) -> None:
        """Update the user's conversational state."""
        query = "UPDATE users SET state = ?, state_data = ? WHERE id = ?"
        self._conn.execute(query, (state, state_data, user_id))

    def delete(self, user_id: int) -> None:
        """Delete a user record."""
        query = "DELETE FROM users WHERE id = ?"
        self._conn.execute(query, (user_id,))

    def _row_to_model(self, row: sqlite3.Row) -> User:
        """Convert a SQLite row to a User model."""
        data = dict(row)

        # Convert created_at from string to datetime
        if isinstance(data["created_at"], str):
            try:
                data["created_at"] = datetime.fromisoformat(data["created_at"]).replace(tzinfo=UTC)
            except ValueError:
                # Handle legacy or potential non-iso formats if necessary
                # For now assuming standard sqlite datetime/timestamp format "YYYY-MM-DD HH:MM:SS"
                data["created_at"] = datetime.strptime(
                    data["created_at"],
                    "%Y-%m-%d %H:%M:%S",
                ).replace(tzinfo=UTC)

        return User(**data)

__init__(conn)

Initialize the UsersTable with a database connection.

Source code in src/loglife/app/db/tables/users.py 
29
30
31
def __init__(self, conn: sqlite3.Connection) -> None:
    """Initialize the UsersTable with a database connection."""
    self._conn = conn

create(phone_number, timezone, referred_by_id=None)

Create a new user record.

Source code in src/loglife/app/db/tables/users.py 
57
58
59
60
61
62
63
64
65
66
67
68
def create(
    self,
    phone_number: str,
    timezone: str,
    referred_by_id: int | None = None,
) -> User:
    """Create a new user record."""
    query = "INSERT INTO users (phone_number, timezone, referred_by_id) VALUES (?, ?, ?)"
    cursor = self._conn.execute(query, (phone_number, timezone, referred_by_id))
    # We need to fetch the created user to return the full model with ID and timestamps
    # The result of get() will not be None here because we just inserted it.
    return self.get(cursor.lastrowid)  # type: ignore[arg-type]

delete(user_id)

Delete a user record.

Source code in src/loglife/app/db/tables/users.py 
112
113
114
115
def delete(self, user_id: int) -> None:
    """Delete a user record."""
    query = "DELETE FROM users WHERE id = ?"
    self._conn.execute(query, (user_id,))

get(user_id)

Retrieve a user by their ID.

Source code in src/loglife/app/db/tables/users.py 
33
34
35
36
37
38
39
40
def get(self, user_id: int) -> User | None:
    """Retrieve a user by their ID."""
    query = "SELECT * FROM users WHERE id = ?"
    row = self._conn.execute(query, (user_id,)).fetchone()

    if row:
        return self._row_to_model(row)
    return None

get_all()

Retrieve all users.

Source code in src/loglife/app/db/tables/users.py 
51
52
53
54
55
def get_all(self) -> list[User]:
    """Retrieve all users."""
    query = "SELECT * FROM users"
    rows = self._conn.execute(query).fetchall()
    return [self._row_to_model(row) for row in rows]

get_by_phone(phone_number)

Retrieve a user by their phone number.

Source code in src/loglife/app/db/tables/users.py 
42
43
44
45
46
47
48
49
def get_by_phone(self, phone_number: str) -> User | None:
    """Retrieve a user by their phone number."""
    query = "SELECT * FROM users WHERE phone_number = ?"
    row = self._conn.execute(query, (phone_number,)).fetchone()

    if row:
        return self._row_to_model(row)
    return None

set_state(user_id, state, state_data=None)

Update the user's conversational state.

Source code in src/loglife/app/db/tables/users.py 
107
108
109
110
def set_state(self, user_id: int, state: str | None, state_data: str | None = None) -> None:
    """Update the user's conversational state."""
    query = "UPDATE users SET state = ?, state_data = ? WHERE id = ?"
    self._conn.execute(query, (state, state_data, user_id))

update(user_id, phone_number=None, timezone=None, send_transcript_file=None, referred_by_id=None)

Update a user record with provided fields.

Source code in src/loglife/app/db/tables/users.py 
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def update(
    self,
    user_id: int,
    phone_number: str | None = None,
    timezone: str | None = None,
    send_transcript_file: int | None = None,
    referred_by_id: int | None = None,
) -> User | None:
    """Update a user record with provided fields."""
    updates = []
    params = []

    if phone_number is not None:
        updates.append("phone_number = ?")
        params.append(phone_number)

    if timezone is not None:
        updates.append("timezone = ?")
        params.append(timezone)

    if send_transcript_file is not None:
        updates.append("send_transcript_file = ?")
        params.append(send_transcript_file)

    if referred_by_id is not None:
        updates.append("referred_by_id = ?")
        params.append(referred_by_id)

    if not updates:
        return self.get(user_id)

    params.append(user_id)
    query = f"UPDATE users SET {', '.join(updates)} WHERE id = ?"
    self._conn.execute(query, params)

    return self.get(user_id)

loglife.app.db.tables.goals

Goal table operations and data model.

This module defines the Goal data class and the GoalsTable class for handling database interactions related to user goals.

Goal dataclass

Goal data model.

Source code in src/loglife/app/db/tables/goals.py 
12
13
14
15
16
17
18
19
20
21
22
@dataclass
class Goal:
    """Goal data model."""

    id: int
    user_id: int
    goal_emoji: str
    goal_description: str
    boost_level: int
    created_at: datetime
    reminder_time: time | None

GoalsTable

Handles database operations for the user_goals table.

Source code in src/loglife/app/db/tables/goals.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class GoalsTable:
    """Handles database operations for the user_goals table."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        """Initialize the GoalsTable with a database connection."""
        self._conn = conn

    def get(self, goal_id: int) -> Goal | None:
        """Retrieve a goal by its ID."""
        query = "SELECT * FROM user_goals WHERE id = ?"
        row = self._conn.execute(query, (goal_id,)).fetchone()
        return self._row_to_model(row) if row else None

    def get_by_user(self, user_id: int) -> list[Goal]:
        """Retrieve all goals for a user."""
        query = "SELECT * FROM user_goals WHERE user_id = ? ORDER BY created_at DESC"
        rows = self._conn.execute(query, (user_id,)).fetchall()
        return [self._row_to_model(row) for row in rows]

    def get_all_with_reminders(self) -> list[Goal]:
        """Retrieve all goals that have a reminder set."""
        query = "SELECT * FROM user_goals WHERE reminder_time IS NOT NULL"
        rows = self._conn.execute(query).fetchall()
        return [self._row_to_model(row) for row in rows]

    def create(
        self,
        user_id: int,
        goal_emoji: str,
        goal_description: str,
        boost_level: int = 1,
    ) -> Goal:
        """Create a new goal record."""
        query = """
            INSERT INTO user_goals(user_id, goal_emoji, goal_description, boost_level)
            VALUES (?, ?, ?, ?)
        """
        cursor = self._conn.execute(query, (user_id, goal_emoji, goal_description, boost_level))
        return self.get(cursor.lastrowid)

    def update(
        self,
        goal_id: int,
        goal_emoji: str | None = None,
        goal_description: str | None = None,
        boost_level: int | None = None,
        reminder_time: str | None = None,
    ) -> Goal | None:
        """Update a goal record with provided fields."""
        updates = []
        params = []

        if goal_emoji is not None:
            updates.append("goal_emoji = ?")
            params.append(goal_emoji)

        if goal_description is not None:
            updates.append("goal_description = ?")
            params.append(goal_description)

        if boost_level is not None:
            updates.append("boost_level = ?")
            params.append(boost_level)

        if reminder_time is not None:
            updates.append("reminder_time = ?")
            params.append(reminder_time)

        if not updates:
            return self.get(goal_id)

        params.append(goal_id)
        query = f"UPDATE user_goals SET {', '.join(updates)} WHERE id = ?"
        self._conn.execute(query, params)

        return self.get(goal_id)

    def delete(self, goal_id: int) -> None:
        """Delete a goal record."""
        query = "DELETE FROM user_goals WHERE id = ?"
        self._conn.execute(query, (goal_id,))

    def _row_to_model(self, row: sqlite3.Row) -> Goal:
        """Convert a SQLite row to a Goal model."""
        data = dict(row)

        # Convert created_at from string to datetime
        if isinstance(data["created_at"], str):
            data["created_at"] = datetime.strptime(
                data["created_at"],
                "%Y-%m-%d %H:%M:%S",
            ).replace(tzinfo=UTC)

        # Convert reminder_time from string to time
        if isinstance(data.get("reminder_time"), str):
            # Expected format is HH:MM:SS
            try:
                dt = datetime.strptime(data["reminder_time"], "%H:%M:%S")  # noqa: DTZ007
                data["reminder_time"] = dt.time()
            except ValueError:
                # Fallback or handle parsing error if format is different
                data["reminder_time"] = None

        return Goal(**data)

__init__(conn)

Initialize the GoalsTable with a database connection.

Source code in src/loglife/app/db/tables/goals.py 
28
29
30
def __init__(self, conn: sqlite3.Connection) -> None:
    """Initialize the GoalsTable with a database connection."""
    self._conn = conn

create(user_id, goal_emoji, goal_description, boost_level=1)

Create a new goal record.

Source code in src/loglife/app/db/tables/goals.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
63
def create(
    self,
    user_id: int,
    goal_emoji: str,
    goal_description: str,
    boost_level: int = 1,
) -> Goal:
    """Create a new goal record."""
    query = """
        INSERT INTO user_goals(user_id, goal_emoji, goal_description, boost_level)
        VALUES (?, ?, ?, ?)
    """
    cursor = self._conn.execute(query, (user_id, goal_emoji, goal_description, boost_level))
    return self.get(cursor.lastrowid)

delete(goal_id)

Delete a goal record.

Source code in src/loglife/app/db/tables/goals.py 
102
103
104
105
def delete(self, goal_id: int) -> None:
    """Delete a goal record."""
    query = "DELETE FROM user_goals WHERE id = ?"
    self._conn.execute(query, (goal_id,))

get(goal_id)

Retrieve a goal by its ID.

Source code in src/loglife/app/db/tables/goals.py 
32
33
34
35
36
def get(self, goal_id: int) -> Goal | None:
    """Retrieve a goal by its ID."""
    query = "SELECT * FROM user_goals WHERE id = ?"
    row = self._conn.execute(query, (goal_id,)).fetchone()
    return self._row_to_model(row) if row else None

get_all_with_reminders()

Retrieve all goals that have a reminder set.

Source code in src/loglife/app/db/tables/goals.py 
44
45
46
47
48
def get_all_with_reminders(self) -> list[Goal]:
    """Retrieve all goals that have a reminder set."""
    query = "SELECT * FROM user_goals WHERE reminder_time IS NOT NULL"
    rows = self._conn.execute(query).fetchall()
    return [self._row_to_model(row) for row in rows]

get_by_user(user_id)

Retrieve all goals for a user.

Source code in src/loglife/app/db/tables/goals.py 
38
39
40
41
42
def get_by_user(self, user_id: int) -> list[Goal]:
    """Retrieve all goals for a user."""
    query = "SELECT * FROM user_goals WHERE user_id = ? ORDER BY created_at DESC"
    rows = self._conn.execute(query, (user_id,)).fetchall()
    return [self._row_to_model(row) for row in rows]

update(goal_id, goal_emoji=None, goal_description=None, boost_level=None, reminder_time=None)

Update a goal record with provided fields.

Source code in src/loglife/app/db/tables/goals.py 
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def update(
    self,
    goal_id: int,
    goal_emoji: str | None = None,
    goal_description: str | None = None,
    boost_level: int | None = None,
    reminder_time: str | None = None,
) -> Goal | None:
    """Update a goal record with provided fields."""
    updates = []
    params = []

    if goal_emoji is not None:
        updates.append("goal_emoji = ?")
        params.append(goal_emoji)

    if goal_description is not None:
        updates.append("goal_description = ?")
        params.append(goal_description)

    if boost_level is not None:
        updates.append("boost_level = ?")
        params.append(boost_level)

    if reminder_time is not None:
        updates.append("reminder_time = ?")
        params.append(reminder_time)

    if not updates:
        return self.get(goal_id)

    params.append(goal_id)
    query = f"UPDATE user_goals SET {', '.join(updates)} WHERE id = ?"
    self._conn.execute(query, params)

    return self.get(goal_id)

loglife.app.db.tables.ratings

Rating table operations and data model.

This module defines the Rating data class and the RatingsTable class for handling database interactions related to goal ratings.

Rating dataclass

Rating data model.

Source code in src/loglife/app/db/tables/ratings.py 
13
14
15
16
17
18
19
20
21
22
@dataclass
class Rating:
    """Rating data model."""

    id: int
    user_goal_id: int
    rating: int
    rating_date: datetime
    created_at: datetime
    updated_at: datetime

RatingsTable

Handles database operations for the goal_ratings table.

Source code in src/loglife/app/db/tables/ratings.py 
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
class RatingsTable:
    """Handles database operations for the goal_ratings table."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        """Initialize the RatingsTable with a database connection."""
        self._conn = conn

    def get(self, rating_id: int) -> Rating | None:
        """Retrieve a rating by its ID."""
        query = "SELECT * FROM goal_ratings WHERE id = ?"
        row = self._conn.execute(query, (rating_id,)).fetchone()
        return self._row_to_model(row) if row else None

    def get_by_goal_and_date(self, user_goal_id: int, rating_date: str) -> Rating | None:
        """Retrieve the most recent rating for a goal on a given date."""
        query = """
            SELECT * FROM goal_ratings
            WHERE user_goal_id = ? AND DATE(rating_date) = DATE(?)
            ORDER BY created_at DESC LIMIT 1
        """
        row = self._conn.execute(query, (user_goal_id, rating_date)).fetchone()
        return self._row_to_model(row) if row else None

    def get_all(self) -> list[Rating]:
        """Retrieve all ratings."""
        query = "SELECT * FROM goal_ratings ORDER BY created_at DESC"
        rows = self._conn.execute(query).fetchall()
        return [self._row_to_model(row) for row in rows]

    def create(self, user_goal_id: int, rating: int) -> Rating:
        """Create a new rating record."""
        query = "INSERT INTO goal_ratings(user_goal_id, rating) VALUES (?, ?)"
        cursor = self._conn.execute(query, (user_goal_id, rating))
        return self.get(cursor.lastrowid)  # type: ignore[arg-type]

    def update(
        self,
        rating_id: int,
        user_goal_id: int | None = None,
        rating: int | None = None,
        rating_date: str | None = None,
    ) -> Rating | None:
        """Update a rating record with provided fields."""
        updates = []
        params = []

        if user_goal_id is not None:
            updates.append("user_goal_id = ?")
            params.append(user_goal_id)

        if rating is not None:
            updates.append("rating = ?")
            params.append(rating)

        if rating_date is not None:
            updates.append("rating_date = ?")
            params.append(rating_date)

        if not updates:
            return self.get(rating_id)

        params.append(rating_id)
        query = f"UPDATE goal_ratings SET {', '.join(updates)} WHERE id = ?"
        self._conn.execute(query, params)

        return self.get(rating_id)

    def delete(self, rating_id: int) -> None:
        """Delete a rating record."""
        query = "DELETE FROM goal_ratings WHERE id = ?"
        self._conn.execute(query, (rating_id,))

    def _row_to_model(self, row: sqlite3.Row) -> Rating:
        """Convert a SQLite row to a Rating model."""
        data = dict(row)

        # Convert timestamps
        for field in ("created_at", "updated_at", "rating_date"):
            if isinstance(data.get(field), str):
                with contextlib.suppress(ValueError):
                    # Try ISO format (handles YYYY-MM-DD and YYYY-MM-DD HH:MM:SS)
                    data[field] = datetime.fromisoformat(data[field]).replace(tzinfo=UTC)

        return Rating(**data)

__init__(conn)

Initialize the RatingsTable with a database connection.

Source code in src/loglife/app/db/tables/ratings.py 
28
29
30
def __init__(self, conn: sqlite3.Connection) -> None:
    """Initialize the RatingsTable with a database connection."""
    self._conn = conn

create(user_goal_id, rating)

Create a new rating record.

Source code in src/loglife/app/db/tables/ratings.py 
54
55
56
57
58
def create(self, user_goal_id: int, rating: int) -> Rating:
    """Create a new rating record."""
    query = "INSERT INTO goal_ratings(user_goal_id, rating) VALUES (?, ?)"
    cursor = self._conn.execute(query, (user_goal_id, rating))
    return self.get(cursor.lastrowid)  # type: ignore[arg-type]

delete(rating_id)

Delete a rating record.

Source code in src/loglife/app/db/tables/ratings.py 
92
93
94
95
def delete(self, rating_id: int) -> None:
    """Delete a rating record."""
    query = "DELETE FROM goal_ratings WHERE id = ?"
    self._conn.execute(query, (rating_id,))

get(rating_id)

Retrieve a rating by its ID.

Source code in src/loglife/app/db/tables/ratings.py 
32
33
34
35
36
def get(self, rating_id: int) -> Rating | None:
    """Retrieve a rating by its ID."""
    query = "SELECT * FROM goal_ratings WHERE id = ?"
    row = self._conn.execute(query, (rating_id,)).fetchone()
    return self._row_to_model(row) if row else None

get_all()

Retrieve all ratings.

Source code in src/loglife/app/db/tables/ratings.py 
48
49
50
51
52
def get_all(self) -> list[Rating]:
    """Retrieve all ratings."""
    query = "SELECT * FROM goal_ratings ORDER BY created_at DESC"
    rows = self._conn.execute(query).fetchall()
    return [self._row_to_model(row) for row in rows]

get_by_goal_and_date(user_goal_id, rating_date)

Retrieve the most recent rating for a goal on a given date.

Source code in src/loglife/app/db/tables/ratings.py 
38
39
40
41
42
43
44
45
46
def get_by_goal_and_date(self, user_goal_id: int, rating_date: str) -> Rating | None:
    """Retrieve the most recent rating for a goal on a given date."""
    query = """
        SELECT * FROM goal_ratings
        WHERE user_goal_id = ? AND DATE(rating_date) = DATE(?)
        ORDER BY created_at DESC LIMIT 1
    """
    row = self._conn.execute(query, (user_goal_id, rating_date)).fetchone()
    return self._row_to_model(row) if row else None

update(rating_id, user_goal_id=None, rating=None, rating_date=None)

Update a rating record with provided fields.

Source code in src/loglife/app/db/tables/ratings.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def update(
    self,
    rating_id: int,
    user_goal_id: int | None = None,
    rating: int | None = None,
    rating_date: str | None = None,
) -> Rating | None:
    """Update a rating record with provided fields."""
    updates = []
    params = []

    if user_goal_id is not None:
        updates.append("user_goal_id = ?")
        params.append(user_goal_id)

    if rating is not None:
        updates.append("rating = ?")
        params.append(rating)

    if rating_date is not None:
        updates.append("rating_date = ?")
        params.append(rating_date)

    if not updates:
        return self.get(rating_id)

    params.append(rating_id)
    query = f"UPDATE goal_ratings SET {', '.join(updates)} WHERE id = ?"
    self._conn.execute(query, params)

    return self.get(rating_id)

loglife.app.db.tables.audio_journals

Audio journal table operations and data model.

This module defines the AudioJournalEntry data class and the AudioJournalsTable class for handling database interactions related to audio journal entries.

AudioJournalEntry dataclass

Audio journal entry data model.

Source code in src/loglife/app/db/tables/audio_journals.py 
12
13
14
15
16
17
18
19
20
@dataclass
class AudioJournalEntry:
    """Audio journal entry data model."""

    id: int
    user_id: int
    transcription_text: str | None
    summary_text: str | None
    created_at: datetime

AudioJournalsTable

Handles database operations for the audio_journals table.

Source code in src/loglife/app/db/tables/audio_journals.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
class AudioJournalsTable:
    """Handles database operations for the audio_journals table."""

    def __init__(self, conn: sqlite3.Connection) -> None:
        """Initialize the AudioJournalsTable with a database connection."""
        self._conn = conn

    def get(self, entry_id: int) -> AudioJournalEntry | None:
        """Retrieve an audio journal entry by its ID."""
        query = "SELECT * FROM audio_journals WHERE id = ?"
        row = self._conn.execute(query, (entry_id,)).fetchone()
        return self._row_to_model(row) if row else None

    def get_by_user(self, user_id: int) -> list[AudioJournalEntry]:
        """Retrieve all audio journal entries for a user."""
        query = """
            SELECT * FROM audio_journals
            WHERE user_id = ?
            ORDER BY created_at DESC
        """
        rows = self._conn.execute(query, (user_id,)).fetchall()
        return [self._row_to_model(row) for row in rows]

    def get_all(self) -> list[AudioJournalEntry]:
        """Retrieve all audio journal entries."""
        query = "SELECT * FROM audio_journals ORDER BY created_at DESC"
        rows = self._conn.execute(query).fetchall()
        return [self._row_to_model(row) for row in rows]

    def create(self, user_id: int, transcription_text: str, summary_text: str) -> None:
        """Create a new audio journal entry."""
        query = """
            INSERT INTO audio_journals (user_id, transcription_text, summary_text)
            VALUES (?, ?, ?)
        """
        self._conn.execute(query, (user_id, transcription_text, summary_text))

    def update(self, entry_id: int, transcription_text: str, summary_text: str) -> None:
        """Update an existing audio journal entry."""
        query = """
            UPDATE audio_journals
            SET transcription_text = ?, summary_text = ?
            WHERE id = ?
        """
        self._conn.execute(query, (transcription_text, summary_text, entry_id))

    def delete(self, entry_id: int) -> None:
        """Delete an audio journal entry."""
        query = "DELETE FROM audio_journals WHERE id = ?"
        self._conn.execute(query, (entry_id,))

    def _row_to_model(self, row: sqlite3.Row) -> AudioJournalEntry:
        """Convert a SQLite row to an AudioJournalEntry model."""
        data = dict(row)

        # Convert created_at from string to datetime
        if isinstance(data.get("created_at"), str):
            data["created_at"] = datetime.strptime(
                data["created_at"],
                "%Y-%m-%d %H:%M:%S",
            ).replace(tzinfo=UTC)

        return AudioJournalEntry(**data)

__init__(conn)

Initialize the AudioJournalsTable with a database connection.

Source code in src/loglife/app/db/tables/audio_journals.py 
26
27
28
def __init__(self, conn: sqlite3.Connection) -> None:
    """Initialize the AudioJournalsTable with a database connection."""
    self._conn = conn

create(user_id, transcription_text, summary_text)

Create a new audio journal entry.

Source code in src/loglife/app/db/tables/audio_journals.py 
52
53
54
55
56
57
58
def create(self, user_id: int, transcription_text: str, summary_text: str) -> None:
    """Create a new audio journal entry."""
    query = """
        INSERT INTO audio_journals (user_id, transcription_text, summary_text)
        VALUES (?, ?, ?)
    """
    self._conn.execute(query, (user_id, transcription_text, summary_text))

delete(entry_id)

Delete an audio journal entry.

Source code in src/loglife/app/db/tables/audio_journals.py 
69
70
71
72
def delete(self, entry_id: int) -> None:
    """Delete an audio journal entry."""
    query = "DELETE FROM audio_journals WHERE id = ?"
    self._conn.execute(query, (entry_id,))

get(entry_id)

Retrieve an audio journal entry by its ID.

Source code in src/loglife/app/db/tables/audio_journals.py 
30
31
32
33
34
def get(self, entry_id: int) -> AudioJournalEntry | None:
    """Retrieve an audio journal entry by its ID."""
    query = "SELECT * FROM audio_journals WHERE id = ?"
    row = self._conn.execute(query, (entry_id,)).fetchone()
    return self._row_to_model(row) if row else None

get_all()

Retrieve all audio journal entries.

Source code in src/loglife/app/db/tables/audio_journals.py 
46
47
48
49
50
def get_all(self) -> list[AudioJournalEntry]:
    """Retrieve all audio journal entries."""
    query = "SELECT * FROM audio_journals ORDER BY created_at DESC"
    rows = self._conn.execute(query).fetchall()
    return [self._row_to_model(row) for row in rows]

get_by_user(user_id)

Retrieve all audio journal entries for a user.

Source code in src/loglife/app/db/tables/audio_journals.py 
36
37
38
39
40
41
42
43
44
def get_by_user(self, user_id: int) -> list[AudioJournalEntry]:
    """Retrieve all audio journal entries for a user."""
    query = """
        SELECT * FROM audio_journals
        WHERE user_id = ?
        ORDER BY created_at DESC
    """
    rows = self._conn.execute(query, (user_id,)).fetchall()
    return [self._row_to_model(row) for row in rows]

update(entry_id, transcription_text, summary_text)

Update an existing audio journal entry.

Source code in src/loglife/app/db/tables/audio_journals.py 
60
61
62
63
64
65
66
67
def update(self, entry_id: int, transcription_text: str, summary_text: str) -> None:
    """Update an existing audio journal entry."""
    query = """
        UPDATE audio_journals
        SET transcription_text = ?, summary_text = ?
        WHERE id = ?
    """
    self._conn.execute(query, (transcription_text, summary_text, entry_id))