Merge branch '2024-sqlalchemy'

2024/sqlalchemy/oop_approach.py

@@ -0,0 +1,31 @@
+import sqlalchemy as sa
+from sqlalchemy.orm import Mapped, mapped_column, sessionmaker, declarative_base
+
+db = sa.create_engine("sqlite:///:memory:")
+Session = sessionmaker(bind=db)
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    username: Mapped[str]
+    email: Mapped[str]
+
+    def __repr__(self) -> str:
+        return f"<User(id={self.id}, username={self.username}, email={self.email})>"
+
+
+def main() -> None:
+    Base.metadata.create_all(db)
+    user = User(username="Arjan", email="[email protected]")
+
+    with Session() as session:
+        session.add(user)
+        session.commit()
+        print(session.query(User).all())
+
+
+if __name__ == "__main__":
+    main()

2024/sqlalchemy/pyproject.toml

@@ -0,0 +1,13 @@
+[tool.poetry]
+name = "sqlalchemy-example"
+version = "0.1.0"
+description = ""
+readme = "README.md"
+
+[tool.poetry.dependencies]
+python = "^3.11"
+SQLAlchemy = "^2.0.27"
+
+[build-system]
+requires = ["poetry-core"]
+build-backend = "poetry.core.masonry.api"

2024/sqlalchemy/relationship.py

@@ -0,0 +1,97 @@
+import hashlib
+import typing
+import sqlalchemy as sa
+from sqlalchemy.orm import (
+    declarative_base,
+    mapped_column,
+    relationship,
+    sessionmaker,
+    Mapped,
+)
+
+db = sa.create_engine("sqlite:///:memory:")
+Session = sessionmaker(bind=db)
+Base = declarative_base()
+
+
+class User(Base):
+    __tablename__ = "users"
+    id: Mapped[int] = mapped_column(primary_key=True)
+    auth: Mapped["UserAuth"] = relationship(
+        "UserAuth", uselist=False, back_populates="user"
+    )
+    posts: Mapped[typing.List["UserPost"]] = relationship(
+        "UserPost", back_populates="user"
+    )
+
+    def __init__(self, username: str, email: str, password: str):
+        super().__init__()
+        self.auth = UserAuth(username=username, email=email)
+        self.auth.set_password(password)
+
+    def __repr__(self) -> str:
+        return f"<User(username={self.auth.username}, email={self.auth.email})>"
+
+
+class UserAuth(Base):
+    __tablename__ = "user_auth"
+
+    id: int = sa.Column(
+        sa.Integer, sa.ForeignKey("users.id"), primary_key=True, index=True, unique=True
+    )
+    username: str = sa.Column(sa.String)
+    email: str = sa.Column(sa.String, unique=True)
+    password_hash: str = sa.Column(sa.String)
+    user: Mapped["User"] = relationship("User", back_populates="auth")
+
+    def __init__(self, username: str, email: str):
+        self.username = username
+        self.email = email
+
+    def set_password(self, password: str) -> None:
+        self.password_hash = hashlib.sha256(password.encode()).hexdigest()
+
+    def check_password(self, password: str) -> bool:
+        return self.password_hash == hashlib.sha256(password.encode()).hexdigest()
+
+    def __repr__(self) -> str:
+        return f"<UserAuth(username={self.username}, email={self.email})>"
+
+
+class UserPost(Base):
+    __tablename__ = "user_posts"
+    id: int = sa.Column(sa.Integer, primary_key=True)
+    user_id: int = sa.Column(
+        sa.Integer, sa.ForeignKey("users.id"), nullable=False, index=True
+    )
+    content: str = sa.Column(sa.String)
+    user: Mapped["User"] = relationship("User", back_populates="posts")
+
+    def __repr__(self) -> str:
+        return f"<UserPost(user={self.user}, content={self.content})>"
+
+
+def main() -> None:
+    Base.metadata.create_all(db)
+
+    with Session.begin() as session:
+        user = User(username="Arjan", email="[email protected]", password="password")
+        post = UserPost(content="Hello World!", user=user)
+        session.add(user)
+        session.add(post)
+
+    with Session.begin() as session:
+        user = session.query(User).first()
+        print(user)
+        print(user.auth)
+        print(user.posts)
+
+        print(f"Password check: {user.auth.check_password('password')}")
+        print(f"Password check: {user.auth.check_password('wrongpassword')}")
+
+        posts = session.query(UserPost).filter(UserPost.user == user).all()
+        print(posts)
+
+
+if __name__ == "__main__":
+    main()

2024/sqlalchemy/simple_approach.py

@@ -0,0 +1,36 @@
+import sqlalchemy as sa
+
+engine = sa.create_engine("sqlite:///:memory:")
+connection = engine.connect()
+
+metadata = sa.MetaData()
+
+user_table = sa.Table(
+    "user",
+    metadata,
+    sa.Column("id", sa.Integer, primary_key=True),
+    sa.Column("username", sa.String),
+    sa.Column("email", sa.String),
+)
+
+
+def insert_user(username: str, email: str) -> None:
+    query = user_table.insert().values(username=username, email=email)
+    connection.execute(query)
+
+
+def select_user(username: str) -> sa.engine.Result:
+    query = user_table.select().where(user_table.c.username == username)
+    result = connection.execute(query)
+    return result.fetchone()
+
+
+def main() -> None:
+    metadata.create_all(engine)
+    insert_user("Arjan", "[email protected]")
+    print(select_user("Arjan"))
+    connection.close()
+
+
+if __name__ == "__main__":
+    main()