SQLAlchemy/JSON Notebook - requires Python 3, SQLAlchemy, psycopg2, and Jupyter (formerly IPython Notebook)

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.

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', 'someone@rawsql.com', '{\"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='use@interpolation.com', 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', 'someone@rawsql.com', {'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=\"use@params.com\",\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', 'use@params.com', {'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', 'use@params.com', {'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=\"me@orms.com\",\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
}