Skip to content

application_vnd_sqlite

Strategy class for application/vnd.sqlite3.

SqliteParseStrategy dataclass

Parse strategy for SQLite.

Registers strategies:

  • ("mediaType", "application/vnd.sqlite3")
Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
@dataclass
@StrategyFactory.register(("mediaType", "application/vnd.sqlite3"))
class SqliteParseStrategy:
    """Parse strategy for SQLite.

    **Registers strategies**:

    - `("mediaType", "application/vnd.sqlite3")`

    """

    resource_config: "ResourceConfig"

    def parse(self, session: "Optional[Dict[str, Any]]" = None) -> "Dict[str, Any]":
        """Parse SQLite query responses."""
        if session is None:
            raise ValueError("Missing session")

        if "sqlquery" in session:
            cn = create_connection(session["filename"])
            cur = cn.cursor()
            rows = cur.execute(session["sqlquery"]).fetchall()
            return {"result": rows}
        return {"result": "No query given"}

    def initialize(
        self, session: "Optional[Dict[str, Any]]" = None
    ) -> "Dict[str, Any]":
        """Initialize."""
        return {}

initialize(self, session=None)

Initialize.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
def initialize(
    self, session: "Optional[Dict[str, Any]]" = None
) -> "Dict[str, Any]":
    """Initialize."""
    return {}

parse(self, session=None)

Parse SQLite query responses.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
def parse(self, session: "Optional[Dict[str, Any]]" = None) -> "Dict[str, Any]":
    """Parse SQLite query responses."""
    if session is None:
        raise ValueError("Missing session")

    if "sqlquery" in session:
        cn = create_connection(session["filename"])
        cur = cn.cursor()
        rows = cur.execute(session["sqlquery"]).fetchall()
        return {"result": rows}
    return {"result": "No query given"}

create_connection(db_file)

create a database connection to the SQLite database specified by db_file :param db_file: database file :return: Connection object or None

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
def create_connection(db_file):
    """create a database connection to the SQLite database
        specified by db_file
    :param db_file: database file
    :return: Connection object or None
    """
    conn = None
    try:
        conn = sqlite3.connect(db_file)
        return conn
    except sqlite3.Error as exc:
        print(exc)

    return conn
Back to top