db

  1import sqlite3
  2import uuid
  3from job import Job
  4
  5
  6def does_exist(job: Job) -> bool:
  7    """Returns True if job exists in database, False otherwise"""
  8
  9    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
 10    c: sqlite3.Cursor = conn.cursor()
 11
 12    c.execute(
 13        "SELECT * FROM jobs WHERE title = ? AND company = ? AND location = ?",
 14        (job.title, job.company, job.location),
 15    )
 16    result = c.fetchone()
 17
 18    conn.close()
 19
 20    return result is not None
 21
 22
 23def edit_job(
 24    job: Job,
 25    field: str,
 26    value: str,
 27) -> None:
 28    """Edits job with specified title"""
 29
 30    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
 31    c: sqlite3.Cursor = conn.cursor()
 32
 33    # Check if job exists in database
 34    if not does_exist(job):
 35        raise ValueError("Job does not exist")
 36
 37    # Check if field is valid
 38    if field not in ("title", "company", "link", "location", "date"):
 39        raise ValueError("Invalid field")
 40
 41    c.execute(f"UPDATE jobs SET {field} = ? WHERE title = ?", (value, job.title))
 42
 43    # Modify job object
 44    match field:
 45        case "title":
 46            job.title = value
 47        case "company":
 48            job.company = value
 49        case "link":
 50            job.link = value
 51        case "location":
 52            job.location = value
 53        case "date":
 54            job.date = value
 55
 56    conn.commit()
 57    conn.close()
 58
 59
 60def delete_job(job: Job) -> None:
 61    """Deletes job with specified title"""
 62
 63    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
 64    c: sqlite3.Cursor = conn.cursor()
 65
 66    # Check if job exists in database
 67    if not does_exist(job):
 68        raise ValueError("Job does not exist")
 69
 70    try:
 71        c.execute("DELETE FROM jobs WHERE title = ?", (job.title,))
 72    except sqlite3.Error as e:
 73        raise e
 74
 75    conn.commit()
 76    conn.close()
 77
 78
 79def get_job(title: str) -> Job:
 80    """Returns job with specified title"""
 81
 82    conn = sqlite3.connect("jobs.db")
 83    c = conn.cursor()
 84
 85    c.execute("SELECT * FROM jobs WHERE title = ?", (title))
 86    result = c.fetchone()
 87
 88    conn.close()
 89
 90    if result is None:
 91        raise ValueError("No job found with that title")
 92
 93    return Job(result[1], result[2], result[3], result[4], result[5], result[6])
 94
 95
 96def get_jobs() -> list[Job]:
 97    """Returns all jobs in database"""
 98
 99    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
100    c: sqlite3.Cursor = conn.cursor()
101
102    c.execute("SELECT * FROM jobs")
103    jobs: list[tuple[int, str, int, str, str, int]] = c.fetchall()
104
105    conn.close()
106
107    return [Job(job[1], job[2], job[3], job[4], job[5], job[6]) for job in jobs]
108
109
110def insert_job(job: Job) -> None:
111    """Inserts job into database"""
112
113    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
114    c: sqlite3.Cursor = conn.cursor()
115
116    # Check if job already exists in database
117    if does_exist(job):
118        return
119
120    try:
121        c.execute(
122            "INSERT INTO jobs VALUES (?, ?, ?, ?, ?, ?, ?)",
123            (
124                (str(uuid.uuid4())),
125                job.title,
126                job.company,
127                job.location,
128                job.link,
129                job.date,
130                job.date_added,
131            ),
132        )
133
134        conn.commit()
135        conn.close()
136    except Exception as e:
137        print(f"Error: {e}")
138
139
140def initialize_database() -> None:
141    """Creates/initializes database"""
142
143    conn: sqlite3.Connection = sqlite3.connect("jobs.db")
144    c: sqlite3.Cursor = conn.cursor()
145
146    # Create table if it does not exist
147    try:
148        c.execute(
149            """CREATE TABLE jobs (
150                    unique_id text primary key,
151                    title text not null,
152                    company text not null,
153                    location text not null,
154                    link text,
155                    date text,
156                    date_added text default current_timestamp
157                )"""
158        )
159    except sqlite3.OperationalError:
160        # Table already exists
161        pass
162
163    conn.commit()
164    conn.close()