Core¶

#| default_exp core

#| export
import sqlalchemy

from fastcore.utils import *
from fastcore.net import urlsave

from collections import namedtuple
from sqlalchemy import create_engine,text,MetaData,Table,Column,engine,sql
from sqlalchemy.sql.base import ReadOnlyColumnCollection
from sqlalchemy.engine.base import Connection
from sqlalchemy.engine.cursor import CursorResult

url = 'https://github.com/lerocha/chinook-database/raw/master/ChinookDatabase/DataSources/Chinook_Sqlite.sqlite'
path = Path('chinook.sqlite')
if not path.exists(): urlsave(url, path)

connstr = f"sqlite:///{path}"

#| export
def conn_db(connstr, **kwargs):
    "Connect to DB using `url.URL()` params and return filled `MetaData`"
    eng = create_engine(connstr, **kwargs)
    conn = eng.connect()
    meta = MetaData()
    meta.reflect(bind=eng)
    meta.bind = eng
    meta.conn = conn
    return meta

db = conn_db(connstr)

#| export
old_md_dir = MetaData.__dir__
old_cc_dir = ReadOnlyColumnCollection.__dir__

@patch
def __dir__(self:MetaData): return old_md_dir(self) + list(self.tables)

@patch
def __dir__(self:ReadOnlyColumnCollection): return old_cc_dir(self) + self.keys()

def _getattr_(self, n):
    if n[0]=='_': raise AttributeError
    if n in self.tables: return self.tables[n]
    #return super().__getattr__(n)
    raise AttributeError
MetaData.__getattr__ = _getattr_

a = db.Album

#| export
@patch
def tuples(self:CursorResult, nm='Row'):
    "Get all results as named tuples"
    rs = self.mappings().fetchall()
    nt = namedtuple(nm, self.keys())
    return [nt(**o) for o in rs]

@patch
def sql(self:Connection, statement, nm='Row', *args, **kwargs):
    "Execute `statement` string and return results (if any)"
    if isinstance(statement,str): statement=text(statement)
    t = self.execute(statement)
    return t.tuples()

@patch
def sql(self:MetaData, statement, *args, **kwargs):
    "Execute `statement` string and return `DataFrame` of results (if any)"
    return self.conn.sql(statement, *args, **kwargs)

#| export
@patch
def get(self:Table, where=None, limit=None):
    "Select from table, optionally limited by `where` and `limit` clauses"
    return self.metadata.conn.sql(self.select().where(where).limit(limit))

#| export
@patch
def close(self:MetaData):
    "Close the connection"
    self.conn.close()

db.close()