Part 1 tutorial demonstartion.

Author: toddbirchard

Makefile

@@ -31,7 +31,7 @@ all help:
 
 .PHONY: run
 run: env
-	service $(PROJECTNAME) start
+	flask run
 
 
 .PHONY: deploy

config.py

@@ -10,4 +10,5 @@
 
 
 SQLALCHEMY_DATABASE_URI = environ.get("SQLALCHEMY_DATABASE_URI")
+SQLALCHEMY_DATABASE_PEM = environ.get("SQLALCHEMY_DATABASE_PEM")
 TIME_NOW = datetime.now()

poetry.lock

@@ -1,3 +1,3 @@
-pymysql==0.10.1
+pymysql==1.0.2; python_version >= "3.6"
 python-dotenv==0.15.0
 sqlalchemy==1.3.22; (python_version >= "2.7" and python_full_version < "3.0.0") or (python_full_version >= "3.4.0")

requirements.txt

@@ -1,8 +1,16 @@
 """Script entry point."""
-from sqlalchemy_tutorial.models import create_tables
+import pprint
 
+from sqlalchemy_tutorial.database.engine import db
+from sqlalchemy_tutorial.queries import fetch_job_listings, update_job_listing
+
+# Print formatter
+pp = pprint.PrettyPrinter(indent=4, width=300)
 
-def init_script():
-    result = create_tables()
-    print(result)
 
+def init_script():
+    """Demonstrate SELECT and UPDATE queries with SQLAlchemy."""
+    rows_selected = fetch_job_listings(db)
+    pp.pprint(rows_selected)
+    rows_updated = update_job_listing(db)
+    print(rows_updated)

sqlalchemy_tutorial/__init__.py

@@ -0,0 +1 @@
+from .engine import db

sqlalchemy_tutorial/database.py

@@ -0,0 +1,15 @@
+"""Create database connection."""
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from config import SQLALCHEMY_DATABASE_PEM, SQLALCHEMY_DATABASE_URI
+
+# Create database engine
+db = create_engine(
+    SQLALCHEMY_DATABASE_URI,
+    connect_args={"ssl": {"key": SQLALCHEMY_DATABASE_PEM}},
+    echo=True,
+)
+
+# Create database session
+session = Session(db)

sqlalchemy_tutorial/database/__init__.py

@@ -1,66 +1,52 @@
 """Declare models and relationships."""
-from sqlalchemy import (
-    Boolean,
-    Column,
-    DateTime,
-    ForeignKey,
-    Integer,
-    String,
-    Table,
-    Text,
-    text,
-)
+from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String, Text
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import relationship
+from sqlalchemy.sql import func
+
 from sqlalchemy_tutorial.database import engine
 
 Base = declarative_base()
 
-association_table = Table(
-    "association",
-    Base.metadata,
-    Column("team_id", Integer, ForeignKey("sport.player.team_id")),
-    Column("id", Integer, ForeignKey("sport.team.id")),
-)
-
 
 class Player(Base):
-    """Individual player object."""
+    """Individual player belonging to a team."""
 
     __tablename__ = "player"
 
     id = Column(Integer, primary_key=True, autoincrement="auto")
-    team_id = Column(Integer, ForeignKey("sport.team.id"), nullable=False)
+    team_id = Column(Integer, ForeignKey("team.id"), nullable=False)
     first_name = Column(String(255), nullable=False)
     last_name = Column(String(255), nullable=False)
     position = Column(String(100), nullable=False)
     injured = Column(Boolean)
     description = Column(Text, nullable=True)
-    created_at = Column(DateTime)
-    updated_at = Column(DateTime, server_default=text("NOW()"), nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, onupdate=func.now())
 
     # Relationships
     team = relationship("Team")
 
     def __repr__(self):
-        return "<Person model {}>".format(self.id)
+        return "<Player {}>".format(self.id)
 
 
 class Team(Base):
-    """Sport team consisting of players."""
+    """Team consisting of many players."""
 
     __tablename__ = "team"
 
     id = Column(Integer, primary_key=True, autoincrement="auto")
     name = Column(String(255), nullable=False)
     city = Column(String(255), nullable=False)
-    created_at = Column(DateTime, nullable=False)
-    updated_at = Column(DateTime, server_default=text("NOW()"), nullable=False)
+    abbreviation = Column(String(255), nullable=False)
+    logo = Column(String(255), nullable=False)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, onupdate=func.now())
 
     def __repr__(self):
-        return "<Team model {}>".format(self.id)
+        return "<Team {}>".format(self.id)
 
 
 def create_tables():
     return Base.metadata.create_all(engine)
-

sqlalchemy_tutorial/database/engine.py

@@ -0,0 +1,28 @@
+"""Executing raw SQL queries against an SQLAlchemy engine."""
+from typing import List, Optional
+
+from sqlalchemy import engine, text
+
+
+def fetch_job_listings(db: engine) -> Optional[List[dict]]:
+    """Select rows from database and parse as list of dicts."""
+    results = db.execute(
+        "SELECT job_id, agency, business_title, \
+        salary_range_from, salary_range_to \
+        FROM nyc_jobs ORDER BY RAND();"
+    )
+    print(f"Selected {results.rowcount} rows.")
+    rows = [dict(row) for row in results.fetchall()]
+    return rows
+
+
+def update_job_listing(db: engine) -> Optional[List[dict]]:
+    """Update row in database with problematic characters escaped."""
+    result = db.execute(
+        text(
+            "UPDATE nyc_jobs SET business_title = 'Senior QA Scapegoat 🏆', \
+            job_category = 'Information? <>!#%%Technology!%%#^&%* & Telecom' \
+            WHERE job_id = 229837;"
+        )
+    )
+    return result.rowcount