Skip to content

application_vnd_sqlite

Strategy class for application/vnd.sqlite3.

SessionUpdateSqLiteParse (SessionUpdate) pydantic-model

Configuration model for SqLiteParse.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
class SessionUpdateSqLiteParse(SessionUpdate):
    """Configuration model for SqLiteParse."""

    result: Optional[list] = Field(None, description="List of results from the query.")
    msg: str = Field(..., description="Messsage concerning the execution of the query.")

msg: str pydantic-field required

Messsage concerning the execution of the query.

result: list pydantic-field

List of results from the query.

SqliteParseStrategy dataclass

Parse strategy for SQLite.

Registers strategies:

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

    **Registers strategies**:

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

    """

    parse_config: "ResourceConfig"

    def get(
        self, session: "Optional[Dict[str, Any]]" = None
    ) -> SessionUpdateSqLiteParse:
        """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 SessionUpdateSqLiteParse(result=rows, msg="Query executed")
        return SessionUpdateSqLiteParse(msg="No query given")

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

get(self, session=None)

Parse SQLite query responses.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
def get(
    self, session: "Optional[Dict[str, Any]]" = None
) -> SessionUpdateSqLiteParse:
    """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 SessionUpdateSqLiteParse(result=rows, msg="Query executed")
    return SessionUpdateSqLiteParse(msg="No query given")

initialize(self, session=None)

Initialize.

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

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