From 2702fa05a1b939e6cfd50c287f52e76157e3cbfe Mon Sep 17 00:00:00 2001 From: Arjan Egges Date: Tue, 5 Mar 2024 14:29:44 +0100 Subject: [PATCH] Updated code example --- 2024/sqlalchemy/oop_approach.py | 18 ++++----- 2024/sqlalchemy/relationship.py | 60 ++++++++++++++++++------------ 2024/sqlalchemy/simple_approach.py | 19 +++++----- 3 files changed, 56 insertions(+), 41 deletions(-) diff --git a/2024/sqlalchemy/oop_approach.py b/2024/sqlalchemy/oop_approach.py index 56197c0c..a4bab7f1 100644 --- a/2024/sqlalchemy/oop_approach.py +++ b/2024/sqlalchemy/oop_approach.py @@ -1,25 +1,25 @@ import sqlalchemy as sa from sqlalchemy.orm import Mapped, mapped_column, sessionmaker, declarative_base -db = sa.create_engine('sqlite:///:memory:') -Session = sessionmaker(bind = db) +db = sa.create_engine("sqlite:///:memory:") +Session = sessionmaker(bind=db) Base = declarative_base() class User(Base): - __tablename__ = 'users' + __tablename__ = "users" - id: Mapped[int] = mapped_column(primary_key = True) - username: str = sa.Column(sa.String) - email: str = sa.Column(sa.String) + id: Mapped[int] = mapped_column(primary_key=True) + username: Mapped[str] + email: Mapped[str] def __repr__(self) -> str: - return f'' + return f"" def main() -> None: Base.metadata.create_all(db) - user = User(username = 'Arjan', email = "Arjan@arjancodes.com") + user = User(username="Arjan", email="Arjan@arjancodes.com") with Session() as session: session.add(user) @@ -27,5 +27,5 @@ def main() -> None: print(session.query(User).all()) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/2024/sqlalchemy/relationship.py b/2024/sqlalchemy/relationship.py index 7b20b28f..13b4027c 100644 --- a/2024/sqlalchemy/relationship.py +++ b/2024/sqlalchemy/relationship.py @@ -1,36 +1,48 @@ 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) +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') + __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 = UserAuth(username=username, email=email) self.auth.set_password(password) def __repr__(self) -> str: - return f'' + return f"" class UserAuth(Base): - __tablename__ = 'user_auth' + __tablename__ = "user_auth" - id: int = sa.Column(sa.Integer, sa.ForeignKey('users.id'), primary_key = True, index = True, unique = True) + 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) + email: str = sa.Column(sa.String, unique=True) password_hash: str = sa.Column(sa.String) - user: Mapped["User"] = relationship('User', back_populates = 'auth') + user: Mapped["User"] = relationship("User", back_populates="auth") def __init__(self, username: str, email: str): self.username = username @@ -43,26 +55,28 @@ def check_password(self, password: str) -> bool: return self.password_hash == hashlib.sha256(password.encode()).hexdigest() def __repr__(self) -> str: - return f'' + return f"" 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) + __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') + user: Mapped["User"] = relationship("User", back_populates="posts") def __repr__(self) -> str: - return f'' + return f"" def main() -> None: Base.metadata.create_all(db) with Session.begin() as session: - user = User(username = 'Arjan', email = "Arjan@arjancodes.com", password = "password") - post = UserPost(content = 'Hello World!', user = user) + user = User(username="Arjan", email="Arjan@arjancodes.com", password="password") + post = UserPost(content="Hello World!", user=user) session.add(user) session.add(post) @@ -79,5 +93,5 @@ def main() -> None: print(posts) -if __name__ == '__main__': +if __name__ == "__main__": main() diff --git a/2024/sqlalchemy/simple_approach.py b/2024/sqlalchemy/simple_approach.py index 5c835616..1b9b5410 100644 --- a/2024/sqlalchemy/simple_approach.py +++ b/2024/sqlalchemy/simple_approach.py @@ -1,24 +1,25 @@ import sqlalchemy as sa -engine = sa.create_engine('sqlite:///:memory:') +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), + "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) +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: +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() @@ -31,5 +32,5 @@ def main() -> None: connection.close() -if __name__ == '__main__': +if __name__ == "__main__": main()