example/postgres-bq.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import dynamic_singer as dsinger\n",
    "from dynamic_singer.extra import postgres\n",
    "from google.cloud import secretmanager\n",
    "import json\n",
    "import psycopg2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/usr/local/lib/python3.7/site-packages/google/auth/_default.py:69: UserWarning: Your application has authenticated using end user credentials from Google Cloud SDK. We recommend that most server applications use service accounts instead. If your application continues to use end user credentials from Cloud SDK, you might receive a \"quota exceeded\" or \"API not enabled\" error. For more information about service accounts, see https://cloud.google.com/docs/authentication/\n",
      "  warnings.warn(_CLOUD_SDK_CREDENTIALS_WARNING)\n"
     ]
    }
   ],
   "source": [
    "# create your own connection\n",
    "def get_connection():\n",
    "    \n",
    "    connection = psycopg2.connect(\n",
    "        user = postgres_user,\n",
    "        password = postgres_password,\n",
    "        host = 'localhost',\n",
    "        port = 5432,\n",
    "        database = postgres_database,\n",
    "    )\n",
    "    return connection\n",
    "\n",
    "connection = get_connection()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Before use postgres.Tap\n",
    "\n",
    "We need to create persistent class to save and pull last primary key state.\n",
    "\n",
    "It must an object with method `pull` and `push`.\n",
    "\n",
    "In this example, we simply write to a local file, but in deployment, it is better we store in GCS / S3 / persistent storage."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Persistent:\n",
    "    def __init__(self, schema, table):\n",
    "        self.filename = f'{schema}-{table}'\n",
    "    \n",
    "    def pull(self):\n",
    "        with open(self.filename) as fopen:\n",
    "            return fopen.read()\n",
    "        \n",
    "    def push(self, data):\n",
    "        with open(self.filename, 'w') as fopen:\n",
    "            fopen.write(data)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "# use your own schema and table from postgres\n",
    "# primary key is much better use column datetime, like created at, or updated at\n",
    "\n",
    "schema = ''\n",
    "table = ''\n",
    "primary_key = ''"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "persistent = Persistent(schema, table)\n",
    "tap = postgres.Tap(schema, table, primary_key, connection, persistent)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## generate bigquery schema\n",
    "\n",
    "If you are not confident with auto-generate schema from dynamic-singer, we can use `bigquery_schema` to generate schema based on table definition."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [],
   "source": [
    "table_schema = postgres.bigquery_schema(schema, table, connection)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [],
   "source": [
    "config = {\n",
    "    'project_id': '',\n",
    "    'dataset_id': 'dynamicsinger',\n",
    "    'validate_records': False,\n",
    "    # 'stream_data': False\n",
    "}\n",
    "\n",
    "with open('bigquery-config.json', 'w') as fopen:\n",
    "    json.dump(config, fopen)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [],
   "source": [
    "source = dsinger.Source(tap, tap_name = 'example', tap_key = primary_key, tap_schema = table_schema)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "source.add('target-bigquery --config bigquery-config.json')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "scrolled": false
   },
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "INFO select * from \"schema\".table where updated_at > '2017-07-25 07:32:40.043588' order by updated_at limit 100\n"
     ]
    }
   ],
   "source": [
    "source.start()"
   ]
  },
  {
   "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.7.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}