Organize repo by series chapter.

Author: toddbirchard

.env.example

@@ -1 +1,2 @@
 SQLALCHEMY_DATABASE_URI=mysql+pymysql://myuser:[email protected]:1234/mydatabase
+SQLALCHEMY_DATABASE_PEM="-----BEGIN CERTIFICATE-----\nghdfigfjvgkjdfvfjkhcvdfjhvfghjbfdvfjshdvjghvfgjvcfjdcvckdjh\n-----END CERTIFICATE-----\n"

Makefile

@@ -31,7 +31,7 @@ all help:
 
 .PHONY: run
 run: env
-	flask run
+	python main.py
 
 
 .PHONY: deploy

README.md

@@ -10,7 +10,12 @@
 
 ![SQLAlchemy Tutorial](https://github.com/hackersandslackers/sqlalchemy-tutorial/blob/master/.github/[email protected]?raw=true)
 
-**Tutorial**: https://hackersandslackers.com/python-database-management-sqlalchemy/
+This repository contains the source code for a four-part tutorial series on SQLAlchemy:
+
+1. [Databases in Python Made Easy with SQLAlchemy](https://hackersandslackers.com/python-database-management-sqlalchemy)
+2. [Implement an ORM with SQLAlchemy](https://hackersandslackers.com/implement-sqlalchemy-orm)
+3. [Relationships in SQLAlchemy Data Models](https://hackersandslackers.com/sqlalchemy-data-models)
+4. [Constructing Database Queries with SQLAlchemy](https://hackersandslackers.com/database-queries-sqlalchemy-orm)
 
 # Getting Started
 
@@ -22,6 +27,7 @@ Replace the values in **.env.example** with your values and rename this file to
 
 
 * `SQLALCHEMY_DATABASE_URI`: Connection URI of a SQL database.
+* `SQLALCHEMY_DATABASE_PEM` _(Optional)_: PEM key for databases requiring an SSL connection.
 
 *Remember never to commit secrets saved in .env files to Github.*
 

config.py

@@ -1,5 +1,4 @@
 """Database config."""
-from datetime import datetime
 from os import environ, path
 
 from dotenv import load_dotenv
@@ -8,7 +7,6 @@
 basedir = path.abspath(path.dirname(__file__))
 load_dotenv(path.join(basedir, ".env"))
 
-
+# Database connection variables
 SQLALCHEMY_DATABASE_URI = environ.get("SQLALCHEMY_DATABASE_URI")
 SQLALCHEMY_DATABASE_PEM = environ.get("SQLALCHEMY_DATABASE_PEM")
-TIME_NOW = datetime.now()

sqlalchemy_tutorial/database/__init__.py renamed to database/__init__.py

@@ -1,4 +1,4 @@
-"""Create database connection."""
+"""Create SQLAlchemy engine and session objects."""
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
 

sqlalchemy_tutorial/database/connect.py renamed to database/connect.py

@@ -1,3 +1,4 @@
+"""Script entry point."""
 from sqlalchemy_tutorial import init_script
 
 if __name__ == "__main__":

main.py

@@ -22,6 +22,7 @@ python = "^3.8"
 sqlalchemy = "*"
 pymysql = "*"
 python-dotenv = "*"
+loguru = "^0.5.3"
 
 [tool.poetry.dev-dependencies]
 pytest = "*"

poetry.lock

@@ -1,3 +1,6 @@
+colorama==0.4.4; python_version >= "3.5" and python_full_version < "3.0.0" and sys_platform == "win32" or sys_platform == "win32" and python_version >= "3.5" and python_full_version >= "3.5.0"
+loguru==0.5.3; python_version >= "3.5"
 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")
+win32-setctime==1.0.3; sys_platform == "win32" and python_version >= "3.5"

pyproject.toml

@@ -1,21 +1,20 @@
-"""Script entry point."""
-import pprint
+"""Run source code for each tutorial found on HackersAndSlackers."""
+from database import db, session
 
-from sqlalchemy_tutorial.database.connect import db
-from sqlalchemy_tutorial.database.models import create_tables
-from sqlalchemy_tutorial.orm import orm_create_user
-from sqlalchemy_tutorial.queries import fetch_job_listings, update_job_listing
-
-# Print formatter
-pp = pprint.PrettyPrinter(indent=4, width=300)
+from .logger import LOGGER
+from .part1_connections import fetch_job_listings, update_job_listing
+from .part2_orm_models import orm_create_user, orm_delete_user
 
 
 def init_script():
-    """Demonstrate SELECT and UPDATE queries with SQLAlchemy."""
-    create_tables()
-    rows_selected = fetch_job_listings(db)
-    pp.pprint(rows_selected)
-    rows_updated = update_job_listing(db)
-    print(rows_updated)
-    user = orm_create_user()
-    print(user)
+    """Run all scripts."""
+
+    # Part 1: Executing SELECT and UPDATE queries with an SQLAlchemy engine
+    LOGGER.info("Part 1: Executing queries against an SQLAlchemy engine.")
+    fetch_job_listings(db)
+    update_job_listing(db)
+
+    # Part 2: Implementing an ORM
+    LOGGER.info("Part 2: Create and delete records from a data model.")
+    orm_create_user(session)
+    orm_delete_user(session)

requirements.txt

@@ -0,0 +1,21 @@
+"""Custom logger."""
+from sys import stdout
+
+from loguru import logger as custom_logger
+
+
+def create_logger() -> custom_logger:
+    """Create custom logger."""
+    custom_logger.remove()
+    custom_logger.add(
+        stdout,
+        colorize=True,
+        catch=True,
+        format="<light-cyan>{time:MM-DD-YYYY HH:mm:ss}</light-cyan> | "
+        + "<light-green>{level}</light-green>: "
+        + "<light-white>{message}</light-white>",
+    )
+    return custom_logger
+
+
+LOGGER = create_logger()

sqlalchemy_tutorial/__init__.py

@@ -0,0 +1 @@
+from .engines import fetch_job_listings, update_job_listing

sqlalchemy_tutorial/logger.py

@@ -1,28 +1,51 @@
-"""Executing raw SQL queries against an SQLAlchemy engine."""
+"""Execute raw SQL queries against an SQLAlchemy engine."""
 from typing import List, Optional
 
 from sqlalchemy import engine, text
 
+from sqlalchemy_tutorial.logger import LOGGER
+
 
 def fetch_job_listings(db: engine) -> Optional[List[dict]]:
-    """Select rows from database and parse as list of dicts."""
+    """
+    Select rows from database and parse as list of dicts.
+
+    :param db: Engine object representing a SQL database.
+    :type db: engine
+
+    :returns: Optional[List[dict]]
+    """
     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()]
+    LOGGER.info(
+        f"Selected {results.rowcount} rows: \
+        {rows}"
+    )
     return rows
 
 
 def update_job_listing(db: engine) -> Optional[List[dict]]:
-    """Update row in database with problematic characters escaped."""
+    """
+    Update row in database with problematic characters escaped.
+
+    :param db: Engine object representing a SQL database.
+    :type db: engine
+
+    :returns: Optional[List[dict]]
+    """
     result = db.execute(
         text(
             "UPDATE nyc_jobs SET business_title = 'Senior QA Scapegoat 🏆', \
             job_category = 'Information? <>!#%%Technology!%%#^&%* & Telecom' \
             WHERE job_id = 229837;"
         )
     )
+    LOGGER.info(
+        f"Selected {result.rowcount} row: \
+        {result}"
+    )
     return result.rowcount

sqlalchemy_tutorial/orm.py

@@ -0,0 +1 @@
+from .orm import orm_create_user, orm_delete_user

sqlalchemy_tutorial/part1_connections/__init__.py

@@ -0,0 +1,32 @@
+"""Declare models and relationships."""
+from sqlalchemy import Column, DateTime, Integer, String, Text
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.sql import func
+
+from database import db
+
+Base = declarative_base()
+
+
+class User(Base):
+    """User account."""
+
+    __tablename__ = "user"
+
+    id = Column(Integer, primary_key=True, autoincrement="auto")
+    username = Column(String(255), unique=True, nullable=False)
+    password = Column(Text, nullable=False)
+    email = Column(String(255), unique=True, nullable=False)
+    first_name = Column(String(255))
+    last_name = Column(String(255))
+    bio = Column(Text)
+    avatar_url = Column(Text)
+    last_seen = Column(DateTime)
+    created_at = Column(DateTime, server_default=func.now())
+    updated_at = Column(DateTime, onupdate=func.now())
+
+    def __repr__(self):
+        return "<User %r>" % self.username
+
+
+Base.metadata.create_all(db)

sqlalchemy_tutorial/queries.py renamed to sqlalchemy_tutorial/part1_connections/engines.py

@@ -0,0 +1,46 @@
+"""Create, delete and update records with SQLAlchemy's ORM."""
+from sqlalchemy.orm import Session
+
+from sqlalchemy_tutorial.logger import LOGGER
+from sqlalchemy_tutorial.part2_orm_models.models import User
+
+# Define instance of our `User` model
+user = User(
+    username="admin",
+    password="Please don't set passwords like this",
+    email="[email protected]",
+    first_name="Todd",
+    last_name="Birchard",
+    bio="I write tutorials on the internet.",
+    avatar_url="https://storage.googleapis.com/hackersandslackers-cdn/authors/[email protected]",
+)
+
+
+def orm_create_user(session: Session):
+    """
+    Create a new instance of our `User` model.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: None
+    """
+    session.add(user)  # Add the user
+    session.commit()  # Commit the change
+
+    LOGGER.info(f"Create new user: {user}")
+
+
+def orm_delete_user(session: Session):
+    """
+    Delete a user if it exists.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: None
+    """
+    session.delete(user)  # Delete the user
+    session.commit()  # Commit the change
+
+    LOGGER.info(f"Deleted existing user: {user}")