looking for job using Linkedin API

main.py

import os
import sqlite3
from linkedin_api import Linkedin
import logging


def extract_id(urn):
    return urn.split(":")[-1]


def search_jobs(api, conn, keywords, location):
    c = conn.cursor()

    jobs = api.search_jobs(
        keywords=keywords,
        location_name=location,
        listed_at=86400 * 2,
        # limit=100,
    )

    new_jobs = 0

    for job in jobs:
        job_id = extract_id(job["dashEntityUrn"])
        job_title = job["title"]

        c.execute("SELECT * FROM jobs WHERE id=?", (job_id,))
        if c.fetchone():
            continue

        new_jobs += 1

        c.execute(
            """INSERT OR IGNORE INTO jobs
            (id, title, location, listedAt, workRemoteAllowed)
            VALUES (?, ?, ?, ?, ?)""",
            (
                job_id,
                job["title"],
                job["formattedLocation"],
                job["listedAt"],
                job["workRemoteAllowed"],
            ),
        )

        try:
            company_id = extract_id(job["companyDetails"]["company"])
        except KeyError:
            logging.error(
                "KeyError: company not found, %s %s", job_id, job_title
            )
            continue

        c.execute(
            """
            INSERT OR IGNORE INTO companies VALUES (?, ?)
        """,
            (company_id, job_id),
        )

    conn.commit()

    logging.info(
        f"{keywords} in {location} found {new_jobs}/{len(jobs)} new jobs"
    )


jobs_table = """
CREATE TABLE IF NOT EXISTS jobs (
    id INTEGER PRIMARY KEY,
    title TEXT,
    location TEXT,
    listedAt INTEGER,
    workRemoteAllowed BOOLEAN,
    isVisible BOOLEAN DEFAULT 1
)
"""

companies_table = """
CREATE TABLE IF NOT EXISTS companies (
    id INTEGER PRIMARY KEY,
    jobId INTEGER,
    FOREIGN KEY(jobId) REFERENCES jobs(id)
)
"""


def main():
    email = os.environ.get("LINKEDIN_EMAIL")
    password = os.environ.get("LINKEDIN_PASSWORD")

    if not email or not password:
        raise Exception(
            "Please provide linkedin credentials, "
            "LINKEDIN_EMAIL and LINKEDIN_PASSWORD"
        )

    conn = sqlite3.connect("linkedin_jobs.db")
    c = conn.cursor()

    c.execute(jobs_table)
    c.execute(companies_table)

    api = Linkedin(email, password)

    locations = [
        "Australia",
        "Canada",
        "Denmark",
        "Finland",
        "Germany",
        "Netherlands",
        "New Zealand",
        "Norway",
        "Sweden",
        "United Kingdom",
        "United States",
    ]

    ignore_keywords = [
        "java", "Ruby", "PHP"
    ]

    search_keywords = ["Python Go", "Python Golang"]

    for keywords in search_keywords:
        for location in locations:
            logging.debug(f"searching jobs {keywords} in {location}")
            search_jobs(api, conn, keywords, location)

            for h in ignore_keywords:
                query = f"UPDATE jobs SET isVisible=0 WHERE title LIKE '%{h}%'"
                c.execute(query)
                conn.commit()

    conn.close()


if __name__ == "__main__":
    LOGLEVEL = os.environ.get("LOGLEVEL", "INFO").upper()
    logging.basicConfig(format="%(asctime)s %(message)s", level=LOGLEVEL)

    main()

requirements.txt

linkedin-api==2.0.1