view env/lib/python3.9/site-packages/galaxy/util/sqlite.py @ 0:4f3585e2f14b draft default tip

"planemo upload commit 60cee0fc7c0cda8592644e1aad72851dec82c959"

import re
import sqlite3

try:
    import sqlparse

    def is_read_only_query(query):
        statements = sqlparse.parse(query)
        for statement in statements:
            if statement.get_type() != "SELECT":
                return False
        return True

except ImportError:
    # Without sqlparse we use a very weak regex check
    def is_read_only_query(query):
        if re.match("select ", query, re.IGNORECASE):
            if re.search("^([^\"]|\"[^\"]*\")*?;", query) or re.search("^([^\']|\'[^\']*\')*?;", query):
                return False
            else:
                return True
        return False


def connect(path):
    connection = sqlite3.connect(path)
    connection.row_factory = sqlite3.Row
    return connection