notebook.community

Edit and run

Smart Drilhole: Prototype DDL with Core

In this section we create a drillhole database using SQLAlchemy. The idea is to reuse the function to create the same database in SQLITE and MySQL.

Restrictions:

  • use PRAGMA foreign_keys = ON; in SQLITE
  • use PRAGMA case_sensitive_like = True; in SQLITE
  • use InnoDB in MySQL




In [1]:



    


import sqlalchemy
sqlalchemy.__version__



    


















    
Out[1]:








'1.1.9'

















In [2]:



    


from sqlalchemy import (create_engine, 
                       Table, 
                       Column, 
                       Float,  
                       String, 
                       MetaData, 
                       ForeignKey, 
                       CheckConstraint)


def og_connect(con_string= 'sqlite:///test2.sqlite', overwrite = False, echo=True): 
    
    print 'connection string:', con_string
    print 'echo', echo
    
    
    # get database type
    dbtype=con_string[0:5]
        
    if dbtype=='sqlit':
        
        # TODO: check database is empty
        
        # Do a row connection and update some pragma
        eng = create_engine(con_string, echo=echo)
        with eng.connect() as con:
            print con.execute('PRAGMA foreign_keys = ON;') 
            print con.execute('PRAGMA case_sensitive_like = True;')
            
    if dbtype=='postg':
        eng = create_engine(con_string, echo=echo)
    
    if dbtype=='sqlite':
        pass
    
    #create collar table
    meta = MetaData()
    collar = Table('collar', meta,
                    Column('BHID', String, primary_key=True),
                    Column('xcollar', Float, nullable=False),
                    Column('ycollar', Float, nullable=False),
                    Column('zcollar', Float, nullable=False),
                    Column('LENGTH', Float, nullable=False),
                    Column('Comments', String))
    
    survey = Table('survey', meta,
                    Column('BHID', None, 
                           ForeignKey(column='collar.BHID', 
                             ondelete='CASCADE', 
                             onupdate='CASCADE',
                             name = 'chk_bhid'),
                           primary_key=True),
                    Column('at', Float, nullable=False, primary_key=True),
                    Column('az', Float, nullable=False),
                    Column('dip', Float, nullable=False),
                    Column('Comments', String))
    
    
    return eng, meta

def og_add_interval(eng, meta, table_name, cols={}):
       
    # create interval table       
    interval = Table(table_name, meta,
                    Column('BHID', None, 
                           ForeignKey(column='collar.BHID', 
                             ondelete='CASCADE', 
                             onupdate='CASCADE',
                             name = 'chk_bhid'),
                           primary_key=True),
                    Column('FROM', Float, nullable=False, primary_key=True),
                    Column('TO', Float, nullable=False),
                    Column('Comments', String),
                    CheckConstraint('"TO" > "FROM"', name='check_interv'))
    
    for col in cols:
        if 'foreignkey' in cols[col]:
            fk = ForeignKey (column= cols[col]['foreignkey']['column'],
                             ondelete= cols[col]['foreignkey']['ondelete'],
                             onupdate= cols[col]['foreignkey']['onupdate'])
            tmpcol = Column(col, None, fk)
        else:
            tmpcol = Column(col, cols[col]['coltypes'], nullable=cols[col]['nullable'])
        
        interval.append_column(tmpcol)

def og_references(eng, meta, table_name = 'assay_certificate', key = 'SampleID', cols={}):
       
    # create interval table       
    interval = Table(table_name, meta,
                    Column(key, String, primary_key=True),
                    Column('Comments', String)) 
    for col in cols:
        tmpcol = Column(col, cols[col]['coltypes'], nullable=cols[col]['nullable'])
        interval.append_column(tmpcol)
    

# TODO: add some fuctions to activate/desactivate constraints 0
# TODO: implement some triggers compatible (see dialects)???
    
def execute(eng, meta):
    meta.create_all(eng)



    











In [3]:



    


#con_string= 'sqlite:///test2.sqlite'
con_string= 'postgresql://postgres@localhost/Dhole'

eng, meta = og_connect(con_string, overwrite = False)



    


















    






connection string: postgresql://postgres@localhost/Dhole
echo True

















In [4]:



    


og_references(eng, meta, table_name = 'assay_certificate', key = 'SampleID', cols={'Au':{'coltypes':Float, 'nullable': True}})
og_references(eng, meta, table_name = 'rock_catalog', key = 'RockID', cols={'Description':{'coltypes':String, 'nullable': True}})
og_add_interval(eng, meta, table_name = 'assay',cols={'SampleID':{'coltypes':String, 
                                                                  'nullable': False,
                                                                  'foreignkey':{'column':'assay_certificate.SampleID',
                                                                                'ondelete':'RESTRICT',
                                                                                'onupdate':'CASCADE'}}})
og_add_interval(eng, meta, table_name = 'litho',cols={'RockID':{'coltypes':String, 
                                                                'nullable': True,
                                                                'foreignkey':{'column':'rock_catalog.RockID',
                                                                                'ondelete':'RESTRICT',
                                                                                'onupdate':'CASCADE'}}})



    











In [5]:



    


execute(eng, meta)



    


















    






2017-05-12 00:43:14,982 INFO sqlalchemy.engine.base.Engine select version()
2017-05-12 00:43:14,983 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:14,986 INFO sqlalchemy.engine.base.Engine select current_schema()
2017-05-12 00:43:14,986 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:14,989 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
2017-05-12 00:43:14,990 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:14,992 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
2017-05-12 00:43:14,993 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:14,993 INFO sqlalchemy.engine.base.Engine show standard_conforming_strings
2017-05-12 00:43:14,994 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:14,997 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:14,997 INFO sqlalchemy.engine.base.Engine {'name': u'assay'}
2017-05-12 00:43:15,002 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:15,003 INFO sqlalchemy.engine.base.Engine {'name': u'rock_catalog'}
2017-05-12 00:43:15,005 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:15,006 INFO sqlalchemy.engine.base.Engine {'name': u'survey'}
2017-05-12 00:43:15,007 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:15,009 INFO sqlalchemy.engine.base.Engine {'name': u'assay_certificate'}
2017-05-12 00:43:15,010 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:15,010 INFO sqlalchemy.engine.base.Engine {'name': u'litho'}
2017-05-12 00:43:15,013 INFO sqlalchemy.engine.base.Engine select relname from pg_class c join pg_namespace n on n.oid=c.relnamespace where pg_catalog.pg_table_is_visible(c.oid) and relname=%(name)s
2017-05-12 00:43:15,013 INFO sqlalchemy.engine.base.Engine {'name': u'collar'}
2017-05-12 00:43:15,016 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE rock_catalog (
	"RockID" VARCHAR NOT NULL, 
	"Comments" VARCHAR, 
	"Description" VARCHAR, 
	PRIMARY KEY ("RockID")
)


2017-05-12 00:43:15,017 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,086 INFO sqlalchemy.engine.base.Engine COMMIT
2017-05-12 00:43:15,107 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE assay_certificate (
	"SampleID" VARCHAR NOT NULL, 
	"Comments" VARCHAR, 
	"Au" FLOAT, 
	PRIMARY KEY ("SampleID")
)


2017-05-12 00:43:15,108 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,197 INFO sqlalchemy.engine.base.Engine COMMIT
2017-05-12 00:43:15,203 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE collar (
	"BHID" VARCHAR NOT NULL, 
	xcollar FLOAT NOT NULL, 
	ycollar FLOAT NOT NULL, 
	zcollar FLOAT NOT NULL, 
	"LENGTH" FLOAT NOT NULL, 
	"Comments" VARCHAR, 
	PRIMARY KEY ("BHID")
)


2017-05-12 00:43:15,204 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,298 INFO sqlalchemy.engine.base.Engine COMMIT
2017-05-12 00:43:15,313 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE assay (
	"BHID" VARCHAR NOT NULL, 
	"FROM" FLOAT NOT NULL, 
	"TO" FLOAT NOT NULL, 
	"Comments" VARCHAR, 
	"SampleID" VARCHAR, 
	PRIMARY KEY ("BHID", "FROM"), 
	CONSTRAINT check_interv CHECK ("TO" > "FROM"), 
	CONSTRAINT chk_bhid FOREIGN KEY("BHID") REFERENCES collar ("BHID") ON DELETE CASCADE ON UPDATE CASCADE, 
	FOREIGN KEY("SampleID") REFERENCES assay_certificate ("SampleID") ON DELETE RESTRICT ON UPDATE CASCADE
)


2017-05-12 00:43:15,313 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,421 INFO sqlalchemy.engine.base.Engine COMMIT
2017-05-12 00:43:15,426 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE survey (
	"BHID" VARCHAR NOT NULL, 
	at FLOAT NOT NULL, 
	az FLOAT NOT NULL, 
	dip FLOAT NOT NULL, 
	"Comments" VARCHAR, 
	PRIMARY KEY ("BHID", at), 
	CONSTRAINT chk_bhid FOREIGN KEY("BHID") REFERENCES collar ("BHID") ON DELETE CASCADE ON UPDATE CASCADE
)


2017-05-12 00:43:15,427 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,499 INFO sqlalchemy.engine.base.Engine COMMIT
2017-05-12 00:43:15,519 INFO sqlalchemy.engine.base.Engine 
CREATE TABLE litho (
	"BHID" VARCHAR NOT NULL, 
	"FROM" FLOAT NOT NULL, 
	"TO" FLOAT NOT NULL, 
	"Comments" VARCHAR, 
	"RockID" VARCHAR, 
	PRIMARY KEY ("BHID", "FROM"), 
	CONSTRAINT check_interv CHECK ("TO" > "FROM"), 
	CONSTRAINT chk_bhid FOREIGN KEY("BHID") REFERENCES collar ("BHID") ON DELETE CASCADE ON UPDATE CASCADE, 
	FOREIGN KEY("RockID") REFERENCES rock_catalog ("RockID") ON DELETE RESTRICT ON UPDATE CASCADE
)


2017-05-12 00:43:15,519 INFO sqlalchemy.engine.base.Engine {}
2017-05-12 00:43:15,598 INFO sqlalchemy.engine.base.Engine COMMIT

















In [ ]:

Content source: opengeostat/SmartDHOLES

Similar notebooks:

notebook.community | gallery | about