import sqlalchemy
from sqlalchemy import *
import logging; log = logging.getLogger('pvl.verkko.db')
# schema
metadata = MetaData()
dhcp_hosts = Table('dhcp_hosts', metadata,
Column('rowid', Integer, primary_key=True),
# unique
Column('ip', String),
Column('mac', String),
Column('gw', String),
# updated
Column('name', String),
Column('first_seen', DateTime),
Column('last_seen', DateTime),
)
# for ORM models
from sqlalchemy.orm import mapper, sessionmaker
Session = sessionmaker()
class Database (object) :
"""
Our underlying database.
"""
# XXX: alias Tables in?
dhcp_hosts = dhcp_hosts
def __init__ (self, database) :
"""
database - sqlalchemy connection URI
"""
self.engine = create_engine(database,
echo = (log.isEnabledFor(logging.DEBUG)),
)
self.session = Session(bind=self.engine)
# ORM
def query (self, cls) :
return self.session.query(cls)
# SQL
def select (self, query) :
return self.engine.connect().execute(query)
def get (self, query) :
"""
Fetch a single row.
XXX: returning None if not found
"""
return self.select(query).fetchone()