Fixed model relationships, added querying by model.

Author: toddbirchard

database/connect.py

@@ -6,9 +6,7 @@
 
 # Create database engine
 db = create_engine(
-    SQLALCHEMY_DATABASE_URI,
-    connect_args={"ssl": {"key": SQLALCHEMY_DATABASE_PEM}},
-    echo=True,
+    SQLALCHEMY_DATABASE_URI, connect_args={"ssl": {"key": SQLALCHEMY_DATABASE_PEM}}
 )
 
 # Create database session

sqlalchemy_tutorial/__init__.py

@@ -4,6 +4,7 @@
 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
+from .part3_relationships import get_posts, orm_create_data
 
 
 def init_script():
@@ -16,5 +17,10 @@ def init_script():
 
     # 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)
+    user = orm_create_user(session)
+    orm_delete_user(session, user)
+
+    # Part 3: ORM relationships
+    LOGGER.info("Part 3: Utilize relationships to execute JOINs.")
+    orm_create_data(session)
+    get_posts(session)

sqlalchemy_tutorial/part2_orm_models/models.py

@@ -21,6 +21,7 @@ class User(Base):
     last_name = Column(String(255))
     bio = Column(Text)
     avatar_url = Column(Text)
+    role = Column(String(255))
     last_seen = Column(DateTime)
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, onupdate=func.now())

sqlalchemy_tutorial/part2_orm_models/orm.py

@@ -1,22 +1,14 @@
 """Create, delete and update records with SQLAlchemy's ORM."""
+from typing import Optional
+
+from sqlalchemy.exc import IntegrityError
 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):
+def orm_create_user(session: Session) -> Optional[User]:
     """
     Create a new instance of our `User` model.
 
@@ -25,22 +17,38 @@ def orm_create_user(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):
+    try:
+        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]",
+        )
+        session.add(user)  # Add the user
+        session.commit()  # Commit the change
+        LOGGER.info(f"Created new user: {user}")
+        return user
+    except IntegrityError as e:
+        LOGGER.error(e.orig)
+
+
+def orm_delete_user(session: Session, user: User):
     """
     Delete a user if it exists.
 
     :param session: SQLAlchemy database session.
     :type session: Session
+    :param user: User created in the above function.
+    :type user: User
 
     :returns: None
     """
-    session.delete(user)  # Delete the user
-    session.commit()  # Commit the change
-
-    LOGGER.info(f"Deleted existing user: {user}")
+    try:
+        session.delete(user)  # Delete the user
+        session.commit()  # Commit the change
+        LOGGER.info(f"Deleted existing user: {user}")
+    except IntegrityError as e:
+        LOGGER.error(e.orig)

sqlalchemy_tutorial/part3_relationships/__init__.py

@@ -0,0 +1,2 @@
+from .joins import get_posts
+from .orm import orm_create_data

sqlalchemy_tutorial/part3_relationships/joins.py

@@ -0,0 +1,17 @@
+"""Perform JOIN queries on models with relationships."""
+from sqlalchemy.orm import Session
+
+from sqlalchemy_tutorial.logger import LOGGER
+from sqlalchemy_tutorial.part3_relationships.models import Comment, Post, User
+
+
+def get_posts(session: Session):
+    """
+    Fetch posts.
+
+    :returns: None
+    """
+    posts = session.query(Post).all()
+    for post in posts:
+        LOGGER.info(f"Post: {post}")
+        LOGGER.info(f"Post author: {post.author}")

sqlalchemy_tutorial/part3_relationships/models.py

@@ -9,43 +9,69 @@
 Base = declarative_base()
 
 
-class Player(Base):
-    """Individual player belonging to a team."""
+class User(Base):
+    """User account."""
 
-    __tablename__ = "player"
+    __tablename__ = "user"
 
     id = Column(Integer, primary_key=True, autoincrement="auto")
-    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)
+    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)
+    role = Column(String(255))
+    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
+
+
+class Comment(Base):
+    """User-generated comment on a blog post."""
+
+    __tablename__ = "comment"
+
+    id = Column(Integer, primary_key=True, index=True)
+    user_id = Column(Integer, ForeignKey("user.id"))
+    post_id = Column(Integer, index=True)
+    body = Column(Text)
+    upvotes = Column(Integer, default=1)
+    removed = Column(Boolean, default=False)
+    created_at = Column(DateTime, server_default=func.now())
+
     # Relationships
-    team = relationship("Team")
+    user = relationship("User", backref="comments")
 
     def __repr__(self):
-        return "<Player {}>".format(self.id)
+        return "<Comment %r>" % self.id
 
 
-class Team(Base):
-    """Team consisting of many players."""
+class Post(Base):
+    """Blog post/article."""
 
-    __tablename__ = "team"
+    __tablename__ = "post"
 
-    id = Column(Integer, primary_key=True, autoincrement="auto")
-    name = Column(String(255), nullable=False)
-    city = Column(String(255), nullable=False)
-    abbreviation = Column(String(255), nullable=False)
-    logo = Column(String(255), nullable=False)
+    id = Column(Integer, primary_key=True, index=True)
+    author_id = Column(Integer, ForeignKey("user.id"))
+    slug = Column(String(255), nullable=False, unique=True)
+    title = Column(String(255), nullable=False)
+    summary = Column(String(400))
+    feature_image = Column(String(300))
+    body = Column(Text)
+    status = Column(String(255), nullable=False, default="unpublished")
     created_at = Column(DateTime, server_default=func.now())
     updated_at = Column(DateTime, onupdate=func.now())
 
+    # Relationships
+    author = relationship("User", backref="posts")
+
     def __repr__(self):
-        return "<Team {}>".format(self.id)
+        return "<Post %r>" % self.slug
 
 
 Base.metadata.create_all(db)

sqlalchemy_tutorial/part3_relationships/orm.py

@@ -0,0 +1,134 @@
+"""Create records from our data models with SQLAlchemy's ORM."""
+from typing import Tuple
+
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from sqlalchemy_tutorial.logger import LOGGER
+from sqlalchemy_tutorial.part3_relationships.models import Comment, Post, User
+
+
+def orm_create_data(session: Session) -> None:
+    """
+    Populate our database with users, posts, and comments to query on.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: None
+    """
+    create_users(session)
+    create_post(session)
+    create_comment(session)
+    LOGGER.info(f"Finished creating user, post, and comment records.")
+
+
+def create_users(session: Session) -> Tuple[User, User]:
+    """
+    Create a couple of user accounts.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: Tuple[User, User]
+    """
+    admin_user = User(
+        username="toddthebod",
+        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]",
+        role="admin",
+    )
+    regular_user = User(
+        username="obnoxioustroll69",
+        password="Please don't set passwords like this",
+        email="[email protected]",
+        first_name="Chad",
+        last_name="Bowswick",
+        bio="I leave hurtful comments on coding tutorials I find on the internet.",
+        avatar_url="https://storage.googleapis.com/hackersandslackers-cdn/authors/[email protected]",
+    )
+    admin_user = create_new_user(admin_user, session)
+    regular_user = create_new_user(regular_user, session)
+    LOGGER.success(f"Created 2 users: {admin_user} & {regular_user}")
+    return admin_user, regular_user
+
+
+def create_new_user(user: User, session: Session) -> User:
+    """
+    Create a new user if username isn't already taken.
+
+    :param user: New user record to create.
+    :type user: User
+    :param session: SQLAlchemy database session.
+    :type session: Session
+    :return: Optional[User]
+    """
+    try:
+        user_query = session.query(User).filter(User.username == user.username).first()
+        if user_query is None:
+            session.add(user)  # Add the user
+            session.commit()  # Commit the change
+            LOGGER.success(f"Created user: {user}")
+        LOGGER.warning(f"User already exists in database: {user}")
+        return user
+    except IntegrityError as e:
+        LOGGER.error(e.orig)
+
+
+def create_post(session: Session) -> Post:
+    """
+    Create a post authored by `admin_user`.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: Post
+    """
+    try:
+        admin_user = session.query(User).filter(User.username == "toddthebod").first()
+        post = Post(
+            author_id=admin_user.id,
+            slug="fake-post-slug",
+            title="Fake Post Title",
+            summary="A fake post to have some fake comments.",
+            feature_image="https://hackersandslackers-cdn.storage.googleapis.com/2021/01/[email protected]",
+            body="Cheese slices monterey jack cauliflower cheese dolcelatte cheese and wine fromage frais rubber cheese gouda. Rubber cheese cheese and wine cheeseburger cheesy grin paneer paneer taleggio caerphilly. Edam mozzarella.",
+        )
+        session.add(admin_user)  # Add the user
+        session.commit()  # Commit the change
+        LOGGER.success(f"Created post {post} published by user {admin_user}")
+        return post
+    except IntegrityError as e:
+        LOGGER.error(e.orig)
+
+
+def create_comment(session: Session) -> Comment:
+    """
+    Create a comment posted by `regular_user` on `admin_user`'s post.
+
+    :param session: SQLAlchemy database session.
+    :type session: Session
+
+    :returns: Comment
+    """
+    try:
+        regular_user = (
+            session.query(User).filter(User.username == "obnoxioustroll69").first()
+        )
+        post = session.query(Post).filter(Post.id == 1).first()
+        comment = Comment(
+            user_id=regular_user.id,
+            post_id=post.id,
+            body="This post about SQLAlchemy is awful. You didn't even bother to explain how to install Python, which is where I (and so many others) got stuck. Plus, your code doesn't even work!! I cloned your code and it keeps giving me `environment variable` errors... WTF are environment variables?!!?!?",
+            upvotes=2,
+        )
+        session.add(comment)  # Add the Comment
+        session.commit()  # Commit the change
+        LOGGER.success(f"Created comment {comment} posted by user {regular_user}")
+        return comment
+    except IntegrityError as e:
+        LOGGER.error(e.orig)