Skip to content

application_vnd_sqlite

Strategy class for application/vnd.sqlite3.

SqLiteParseContent

Bases: AttrDict

Configuration model for SqLiteParse.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
82
83
84
85
class SqLiteParseContent(AttrDict):
    """Configuration model for SqLiteParse."""

    result: list = Field(..., description="List of results from the query.")

result: list = Field(..., description='List of results from the query.') class-attribute instance-attribute

SqliteConfig

Bases: AttrDict

Configuration data model for SqliteParseStrategy.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class SqliteConfig(AttrDict):
    """Configuration data model for
    [`SqliteParseStrategy`][oteapi.strategies.parse.application_vnd_sqlite.SqliteParseStrategy].
    """

    # Resource config
    downloadUrl: Optional[HostlessAnyUrl] = Field(
        None, description=ResourceConfig.model_fields["downloadUrl"].description
    )
    mediaType: Literal["application/vnd.sqlite3"] = Field(
        "application/vnd.sqlite3",
        description=ResourceConfig.model_fields["mediaType"].description,
    )

    # SQLite parse strategy-specific config
    sqlquery: str = Field("", description="A SQL query string.")
    datacache_config: Optional[DataCacheConfig] = Field(
        None,
        description="Configuration options for the local data cache.",
    )

datacache_config: Optional[DataCacheConfig] = Field(None, description='Configuration options for the local data cache.') class-attribute instance-attribute

downloadUrl: Optional[HostlessAnyUrl] = Field(None, description=ResourceConfig.model_fields['downloadUrl'].description) class-attribute instance-attribute

mediaType: Literal['application/vnd.sqlite3'] = Field('application/vnd.sqlite3', description=ResourceConfig.model_fields['mediaType'].description) class-attribute instance-attribute

sqlquery: str = Field('', description='A SQL query string.') class-attribute instance-attribute

SqliteParseStrategy

Parse strategy for SQLite.

Purpose of this strategy: Download a SQLite database using downloadUrl and run a SQL query on the database to return all relevant rows.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
 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
@dataclass
class SqliteParseStrategy:
    """Parse strategy for SQLite.

    Purpose of this strategy: Download a SQLite database using `downloadUrl` and run a
    SQL query on the database to return all relevant rows.

    """

    parse_config: SqliteParserConfig

    def initialize(self) -> AttrDict:
        """Initialize strategy."""
        return AttrDict()

    def get(self) -> SqLiteParseContent:
        """Parse SQLite query responses."""

        if self.parse_config.configuration.downloadUrl is None:
            raise ValueError("No download URL provided.")

        if self.parse_config.configuration.mediaType != "application/vnd.sqlite3":
            raise ValueError("Invalid media type.")

        # Retrieve SQLite file
        downloader = create_strategy(
            "download", self.parse_config.configuration.model_dump()
        )
        cache_key = downloader.get()["key"]

        cache = DataCache(self.parse_config.configuration.datacache_config)
        with cache.getfile(cache_key, suffix="db") as filename:
            connection = create_connection(filename)
            cursor = connection.cursor()
            result = cursor.execute(self.parse_config.configuration.sqlquery).fetchall()
            connection.close()
        return SqLiteParseContent(result=result)

parse_config: SqliteParserConfig instance-attribute

get()

Parse SQLite query responses.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def get(self) -> SqLiteParseContent:
    """Parse SQLite query responses."""

    if self.parse_config.configuration.downloadUrl is None:
        raise ValueError("No download URL provided.")

    if self.parse_config.configuration.mediaType != "application/vnd.sqlite3":
        raise ValueError("Invalid media type.")

    # Retrieve SQLite file
    downloader = create_strategy(
        "download", self.parse_config.configuration.model_dump()
    )
    cache_key = downloader.get()["key"]

    cache = DataCache(self.parse_config.configuration.datacache_config)
    with cache.getfile(cache_key, suffix="db") as filename:
        connection = create_connection(filename)
        cursor = connection.cursor()
        result = cursor.execute(self.parse_config.configuration.sqlquery).fetchall()
        connection.close()
    return SqLiteParseContent(result=result)

initialize()

Initialize strategy.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
 99
100
101
def initialize(self) -> AttrDict:
    """Initialize strategy."""
    return AttrDict()

SqliteParserConfig

Bases: ParserConfig

SQLite parse strategy resource config.

Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
51
52
53
54
55
56
57
58
59
60
class SqliteParserConfig(ParserConfig):
    """SQLite parse strategy resource config."""

    parserType: Literal["parser/sqlite3"] = Field(
        "parser/sqlite3",
        description=ParserConfig.model_fields["parserType"].description,
    )
    configuration: SqliteConfig = Field(
        ..., description="SQLite parse strategy-specific configuration."
    )

configuration: SqliteConfig = Field(..., description='SQLite parse strategy-specific configuration.') class-attribute instance-attribute

parserType: Literal['parser/sqlite3'] = Field('parser/sqlite3', description=ParserConfig.model_fields['parserType'].description) class-attribute instance-attribute

create_connection(db_file)

Create a database connection to SQLite database.

Parameters:

Name Type Description Default
db_file Path

Full path to SQLite database file.

 required

Raises:

Type Description
Error

If a DB connection cannot be made.

Returns:

Type Description
Connection

Connection object.
Source code in oteapi/strategies/parse/application_vnd_sqlite.py 
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
def create_connection(db_file: Path) -> sqlite3.Connection:
    """Create a database connection to SQLite database.

    Parameters:
        db_file: Full path to SQLite database file.

    Raises:
        sqlite3.Error: If a DB connection cannot be made.

    Returns:
        Connection object.

    """
    try:
        return sqlite3.connect(db_file)
    except sqlite3.Error as exc:
        raise sqlite3.Error("Could not connect to given SQLite DB.") from exc