Skip to content

Instantly share code, notes, and snippets.

@philipvollet

philipvollet/README.md

Forked from ryansb/README.md
Created January 23, 2021 14:24
Show Gist options
  • Save philipvollet/13325bd93ad4cb182d4b2c8ff593dc69 to your computer and use it in GitHub Desktop.
Save philipvollet/13325bd93ad4cb182d4b2c8ff593dc69 to your computer and use it in GitHub Desktop.
Download ZIP
SQLAlchemy/JSON Notebook - requires Python 3, SQLAlchemy, psycopg2, and Jupyter (formerly IPython Notebook)
Raw
README.md

Before running this notebook, run:

pip3 install jupyter SQLAlchemy psycopg2

This will install the notebook server and database drivers needed to run these examples. For more information on installing Jupyter (formerly IPython notebook) see their install guide.

Once you've installed the dependencies, run jupyter notebook and it will open your web browser to the notebook's main page. Then upload this notebook (SQLTest.ipynb) and run it.

Display the source blob
Display the rendered blob
Raw
SQLTest.ipynb
{
"cells": [
{
"cell_type": "code",
"execution_count": 1,
"metadata": {
"collapsed": true
},
"outputs": [],
"source": [
"connection_string = 'postgresql:///example'"
]
},
{
"cell_type": "code",
"execution_count": 2,
"metadata": {
"collapsed": true
},
"outputs": [],
"source": [
"import json\n",
"import sqlalchemy\n",
"\n",
"db = sqlalchemy.create_engine(connection_string)\n",
"engine = db.connect()\n",
"meta = sqlalchemy.MetaData(engine)"
]
},
{
"cell_type": "code",
"execution_count": 3,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"from sqlalchemy import Column, Integer, Text\n",
"from sqlalchemy.dialects.postgresql import JSON, JSONB\n",
"\n",
"sqlalchemy.Table(\"jsontable\", meta,\n",
" Column('id', Integer, primary_key=True),\n",
" Column('name', Text),\n",
" Column('email', Text),\n",
" Column('doc', JSON))\n",
"meta.create_all()"
]
},
{
"cell_type": "code",
"execution_count": 4,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"<sqlalchemy.engine.result.ResultProxy at 0x7f786cca4e48>"
]
},
"execution_count": 4,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"j_table = sqlalchemy.table(\"jsontable\",\n",
" Column('id', Integer),\n",
" Column('name', Text),\n",
" Column('email', Text),\n",
" Column('doc', JSON))\n",
"# drop existing records\n",
"engine.execute(j_table.delete())"
]
},
{
"cell_type": "code",
"execution_count": 5,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"<sqlalchemy.engine.result.ResultProxy at 0x7f787d002d30>"
]
},
"execution_count": 5,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"engine.execute(\"\"\"INSERT INTO jsontable VALUES (1, 'Mr. RawSQL', '[email protected]', '{\"dialect\": \"raw\"}')\"\"\")"
]
},
{
"cell_type": "code",
"execution_count": 6,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"<sqlalchemy.engine.result.ResultProxy at 0x7f788399d6a0>"
]
},
"execution_count": 6,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"query = \"\"\"INSERT INTO jsontable VALUES (%(id)s, %(name)s, %(email)s, %(doc)s)\"\"\"\n",
"engine.execute(query, id=2, name='Mr. Interpolated Query', email='[email protected]', doc=json.dumps({'dialect': 'string interpolation'}))"
]
},
{
"cell_type": "code",
"execution_count": 7,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"s = j_table.select().where(j_table.c.name == \"Mr. RawSQL\")"
]
},
{
"cell_type": "code",
"execution_count": 8,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"(1, 'Mr. RawSQL', '[email protected]', {'dialect': 'raw'})"
]
},
"execution_count": 8,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"res = engine.execute(s)\n",
"res.fetchone()"
]
},
{
"cell_type": "code",
"execution_count": 9,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"'INSERT INTO jsontable (id, name, doc) VALUES (:id, :name, :doc)'"
]
},
"execution_count": 9,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"str(j_table.insert().values(id=3, name=\"Mr. Params\", doc={\"hi\": \"There\"}))"
]
},
{
"cell_type": "code",
"execution_count": 10,
"metadata": {
"collapsed": false
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"INSERT INTO jsontable (id, name, email, doc) VALUES (:id, :name, :email, :doc)\n"
]
}
],
"source": [
"ins = j_table.insert().values(id=3,\n",
" name=\"Mr. Params\",\n",
" email=\"[email protected]\",\n",
" doc={\"dialect\": \"params\",\n",
" \"address\": {\"street\": \"Main St.\", \"zip\": 12345},\n",
" },\n",
" )\n",
"print(str(ins))"
]
},
{
"cell_type": "code",
"execution_count": 11,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"<sqlalchemy.engine.result.ResultProxy at 0x7f788399d780>"
]
},
"execution_count": 11,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"engine.execute(ins)"
]
},
{
"cell_type": "code",
"execution_count": 12,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"(3, 'Mr. Params', '[email protected]', {'dialect': 'params', 'address': {'street': 'Main St.', 'zip': 12345}})"
]
},
"execution_count": 12,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"engine.execute(j_table.select().where(j_table.c.doc['dialect'].astext == 'params')).fetchone()"
]
},
{
"cell_type": "code",
"execution_count": 13,
"metadata": {
"collapsed": false
},
"outputs": [
{
"data": {
"text/plain": [
"(3, 'Mr. Params', '[email protected]', {'dialect': 'params', 'address': {'street': 'Main St.', 'zip': 12345}})"
]
},
"execution_count": 13,
"metadata": {},
"output_type": "execute_result"
}
],
"source": [
"engine.execute(j_table.select().where(j_table.c.doc[('address', 'zip')].cast(sqlalchemy.Integer) == 12345)).fetchone()"
]
},
{
"cell_type": "code",
"execution_count": 14,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"from sqlalchemy.ext.declarative import declarative_base\n",
"from sqlalchemy.orm import sessionmaker\n",
"\n",
"Base = declarative_base()\n",
"class User(Base):\n",
" __tablename__ = 'jsontable'\n",
" id = Column(Integer, primary_key=True)\n",
" name = Column(Text)\n",
" email = Column(Text)\n",
" doc = Column(JSON)\n",
"\n",
"Base.metadata.create_all(engine)\n",
"SessionFactory = sessionmaker(engine)"
]
},
{
"cell_type": "code",
"execution_count": 15,
"metadata": {
"collapsed": false
},
"outputs": [],
"source": [
"session = SessionFactory()\n",
"u = User(\n",
" id=4,\n",
" name=\"Oscar ORM\",\n",
" email=\"[email protected]\",\n",
" doc={\"address\": {\"zip\": 5678, \"street\": \"Cross St.\"}})\n",
"session.add(u)\n",
"session.commit()"
]
},
{
"cell_type": "code",
"execution_count": 16,
"metadata": {
"collapsed": false
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"Mr. Params\n",
"{'dialect': 'params', 'address': {'street': 'Main St.', 'zip': 12345}}\n"
]
}
],
"source": [
"uu = session.query(User).filter(User.id == 3).one()\n",
"print(uu.name)\n",
"print(uu.doc)"
]
},
{
"cell_type": "code",
"execution_count": 17,
"metadata": {
"collapsed": false
},
"outputs": [
{
"name": "stdout",
"output_type": "stream",
"text": [
"{'address': {'street': 'Cross St.', 'zip': 5678}}\n"
]
}
],
"source": [
"uu = session.query(User).filter(\n",
" User.doc[\n",
" ('address', 'zip')\n",
" ].cast(sqlalchemy.Integer) == 5678\n",
").one()\n",
"print(uu.doc)"
]
}
],
"metadata": {
"kernelspec": {
"display_name": "Python 3",
"language": "python",
"name": "python3"
},
"language_info": {
"codemirror_mode": {
"name": "ipython",
"version": 3
},
"file_extension": ".py",
"mimetype": "text/x-python",
"name": "python",
"nbconvert_exporter": "python",
"pygments_lexer": "ipython3",
"version": "3.5.0"
}
},
"nbformat": 4,
"nbformat_minor": 0
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment