03_ORM.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "https://docs.sqlalchemy.org/en/13/orm/tutorial.html"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sqlite3"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "conn = sqlite3.connect('data/sqlite3/test.db')\n",
    "conn.execute('DROP TABLE IF EXISTS user_account')\n",
    "conn.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "from typing import List\n",
    "from typing import Optional\n",
    "from sqlalchemy import ForeignKey\n",
    "from sqlalchemy import String\n",
    "from sqlalchemy.orm import DeclarativeBase\n",
    "from sqlalchemy.orm import Mapped\n",
    "from sqlalchemy.orm import mapped_column\n",
    "from sqlalchemy.orm import relationship\n",
    "from sqlalchemy import create_engine"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "path = ! pwd\n",
    "\n",
    "# windows\n",
    "# path = ! cd\n",
    "engine = create_engine(f'sqlite:///{path[0]}/data/sqlite3/test.db')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Defino modelo"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Base(DeclarativeBase):\n",
    "    pass\n",
    "\n",
    "class User(Base):\n",
    "    __tablename__ = \"user_account\"\n",
    "    id: Mapped[int] = mapped_column(primary_key=True)\n",
    "    name: Mapped[str] = mapped_column(String(30))\n",
    "    fullname: Mapped[Optional[str]]\n",
    "    age: Mapped[Optional[int]]\n",
    "    addresses: Mapped[List[\"Address\"]] = relationship(\n",
    "        back_populates=\"user\", cascade=\"all, delete-orphan\"\n",
    "    )\n",
    "    \n",
    "    def __repr__(self) -> str:\n",
    "        return f\"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})\"\n",
    "\n",
    "class Address(Base):\n",
    "    __tablename__ = \"address\"\n",
    "    id: Mapped[int] = mapped_column(primary_key=True)\n",
    "    email_address: Mapped[str]\n",
    "    user_id: Mapped[int] = mapped_column(ForeignKey(\"user_account.id\"))\n",
    "    user: Mapped[\"User\"] = relationship(back_populates=\"addresses\")\n",
    "\n",
    "    def __repr__(self) -> str:\n",
    "        return f\"Address(id={self.id!r}, email_address={self.email_address!r})\""
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creo la tabla"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Table('user_account', MetaData(), Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False), Column('name', String(length=30), table=<user_account>, nullable=False), Column('fullname', String(), table=<user_account>), Column('age', Integer(), table=<user_account>), schema=None)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# nos crea automaticamente el Schema\n",
    "User.__table__ "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "Base.metadata.create_all(engine)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "pedro = User(\n",
    "    name=\"Pedro\",\n",
    "    fullname=\"Pedro Gomez\",\n",
    "    age=45,\n",
    "    addresses=[Address(email_address=\"pedro@sqlalchemy.org\")],\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "('Pedro',\n",
       " 'Pedro Gomez',\n",
       " [Address(id=None, email_address='pedro@sqlalchemy.org')],\n",
       " 45)"
      ]
     },
     "execution_count": 9,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "pedro.name, pedro.fullname, pedro.addresses, pedro.age"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Creación de la sessión"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {},
   "outputs": [],
   "source": [
    "from sqlalchemy.orm import Session\n",
    "session = Session(engine)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Agregamos usuarios"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {},
   "outputs": [],
   "source": [
    "ed_user = User(name='ed', fullname='Ed Jones', age=45)\n",
    "session.add(ed_user)\n",
    "session.commit()\n",
    "session.close()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {},
   "outputs": [],
   "source": [
    "with Session(engine) as session:\n",
    "    spongebob = User(\n",
    "        name=\"spongebob\",\n",
    "        fullname=\"Spongebob Squarepants\",\n",
    "        age=40,\n",
    "        addresses=[Address(email_address=\"spongebob@sqlalchemy.org\")],\n",
    "    )\n",
    "    sandy = User(\n",
    "        name=\"sandy\",\n",
    "        fullname=\"Sandy Cheeks\",\n",
    "        age=40,\n",
    "        addresses=[\n",
    "            Address(email_address=\"sandy@sqlalchemy.org\"),\n",
    "            Address(email_address=\"sandy@squirrelpower.org\"),\n",
    "        ],\n",
    "    )\n",
    "    patrick = User(name=\"patrick\", fullname=\"Patrick Star\")\n",
    "    session.add_all([spongebob, sandy, patrick])\n",
    "    session.commit()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Query"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "User(id=2, name='spongebob', fullname='Spongebob Squarepants')\n",
      "User(id=3, name='sandy', fullname='Sandy Cheeks')\n"
     ]
    }
   ],
   "source": [
    "from sqlalchemy import select\n",
    "\n",
    "session = Session(engine)\n",
    "\n",
    "stmt = select(User).where(User.name.in_([\"spongebob\", \"sandy\"]))\n",
    "\n",
    "for user in session.scalars(stmt):\n",
    "    print(user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "4"
      ]
     },
     "execution_count": 14,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.query(User).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Agregamos mas usuarios"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Agregamos otros usuarios\n",
    "session.add_all([User(name='wendy', fullname='Wendy Williams', age=35),\n",
    "                 User(name='mary', fullname='Mary Contrary', age=35),\n",
    "                 User(name='john', fullname='John Lennon', age=20),\n",
    "                 User(name='fred', fullname='Fred Flintstone', age=20)])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "8"
      ]
     },
     "execution_count": 16,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.query(User).count()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Session new and dirty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {},
   "outputs": [],
   "source": [
    "juan_user = User(name='juan', fullname='Juan Perez')\n",
    "session.add(juan_user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([User(id=None, name='juan', fullname='Juan Perez')])"
      ]
     },
     "execution_count": 18,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.new"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([])"
      ]
     },
     "execution_count": 19,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.dirty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([])"
      ]
     },
     "execution_count": 21,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.new"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [],
   "source": [
    "juan_user.name = 'Juan'\n",
    "session.add(juan_user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([User(id=9, name='Juan', fullname='Juan Perez')])"
      ]
     },
     "execution_count": 23,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.dirty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 24,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.commit()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 25,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([])"
      ]
     },
     "execution_count": 25,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.dirty"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Volviendo atras (Rollback)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 26,
   "metadata": {},
   "outputs": [],
   "source": [
    "juan_user.name = 'John'"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([User(id=9, name='John', fullname='Juan Perez')])"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.dirty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "metadata": {},
   "outputs": [],
   "source": [
    "fake_user = User(name='fakeuser', fullname='Invalid')\n",
    "session.add(fake_user)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([User(id=None, name='fakeuser', fullname='Invalid')])"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.new"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 30,
   "metadata": {},
   "outputs": [],
   "source": [
    "session.rollback()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([])"
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.dirty"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 32,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "IdentitySet([])"
      ]
     },
     "execution_count": 32,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.new"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Ordenando"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 34,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[User(id=4, name='patrick', fullname='Patrick Star'),\n",
       " User(id=9, name='Juan', fullname='Juan Perez'),\n",
       " User(id=7, name='john', fullname='John Lennon'),\n",
       " User(id=8, name='fred', fullname='Fred Flintstone'),\n",
       " User(id=5, name='wendy', fullname='Wendy Williams'),\n",
       " User(id=6, name='mary', fullname='Mary Contrary'),\n",
       " User(id=2, name='spongebob', fullname='Spongebob Squarepants'),\n",
       " User(id=3, name='sandy', fullname='Sandy Cheeks'),\n",
       " User(id=1, name='ed', fullname='Ed Jones')]"
      ]
     },
     "execution_count": 34,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.query(User).order_by(User.age).all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 36,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4 patrick Patrick Star None\n",
      "9 Juan Juan Perez None\n",
      "7 john John Lennon 20\n",
      "8 fred Fred Flintstone 20\n",
      "5 wendy Wendy Williams 35\n",
      "6 mary Mary Contrary 35\n",
      "2 spongebob Spongebob Squarepants 40\n",
      "3 sandy Sandy Cheeks 40\n",
      "1 ed Ed Jones 45\n"
     ]
    }
   ],
   "source": [
    "for instance in session.query(User).order_by(User.age):\n",
    "    print(instance.id, instance.name, instance.fullname, instance.age)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Agregación"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 37,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(None, 0), (20, 2), (35, 2), (40, 2), (45, 1)]"
      ]
     },
     "execution_count": 37,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "import sqlalchemy\n",
    "\n",
    "session.query(User.age, sqlalchemy.func.count(User.age)).group_by(User.age).all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 38,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(33.57142857142857,)]"
      ]
     },
     "execution_count": 38,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.query(sqlalchemy.func.avg(User.age)).all()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Filtros"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 39,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[User(id=7, name='john', fullname='John Lennon'),\n",
       " User(id=8, name='fred', fullname='Fred Flintstone')]"
      ]
     },
     "execution_count": 39,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "session.query(User).filter(User.age == 20).all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "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.9.17"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}