Skip to content

Instantly share code, notes, and snippets.

@SegFaultAX
Created November 14, 2019 01:03
Show Gist options
  • Save SegFaultAX/520f4257bbe5eacefea6937b8721944a to your computer and use it in GitHub Desktop.
Save SegFaultAX/520f4257bbe5eacefea6937b8721944a to your computer and use it in GitHub Desktop.
Simple bulk inserts with lookups [sqlite] [sql] [python]
import sqlite3
class Column:
def __init__(self, key, column=None):
self.key = key
self.column = column or self.key
@property
def value(self):
return ":" + self.key
class Lookup:
def __init__(self, colspec, table, fk, comparison_column):
self.colspec = colspec
self.table = table
self.fk = fk
self.comparison_column = comparison_column
@property
def column(self):
return self.colspec.column
@property
def value(self):
return f"""(select {self.fk} from {self.table} where {self.comparison_column} = :{self.colspec.key} limit 1)"""
def bulk_insert(cur, rows, table, colspec):
columns = ", ".join(c.column for c in colspec)
values = ", ".join(c.value for c in colspec)
sql = f"""
insert into {table} ({columns}) values ({values});
"""
print(sql)
cur.executemany(sql, rows)
def main():
conn = sqlite3.connect("test.db")
conn.execute("drop table if exists teams")
conn.execute("""
create table if not exists teams (
id integer primary key autoincrement,
name text not null unique
)
""")
conn.execute("drop table if exists escalations")
conn.execute("""
create table if not exists escalations (
id integer primary key autoincrement,
name text not null unique
)
""")
conn.execute("drop table if exists services")
conn.execute("""
create table if not exists services (
id integer primary key autoincrement,
team_id integer not null references teams(id),
escalation_id integer not null references escalations(id),
name text not null
)
""")
teams = [
{
"name": "team1"
},
{
"name": "team2"
}
]
escalations = [
{
"name": "esc1",
},
{
"name": "esc2",
}
]
services = [
{
"name": "service1",
"team": "team1",
"escalation": "esc1",
},
{
"name": "service2",
"team": "team1",
"escalation": "esc2",
},
{
"name": "service3",
"team": "team2",
"escalation": "esc1",
}
]
cur = conn.cursor()
bulk_insert(cur, teams, "teams", [Column("name")])
bulk_insert(cur, escalations, "escalations", [Column("name")])
bulk_insert(cur, services, "services", [
Column("name"),
Lookup(
Column("team", "team_id"), # colspec: name of the column from the CSV
"teams", # table: table to lookup foreign key
"id", # fk: target field on foreign table
"name" # comparison_column: comparison column on foreign table
),
Lookup(
Column("escalation", "escalation_id"), # colspec: name of the column from the CSV
"escalations", # table: table to lookup foreign key
"id", # fk: target field on foreign table
"name" # comparison_column: comparison column on foreign table
),
])
conn.commit()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment