diff --git a/2024/sqlalchemy/oop_approach.py b/2024/sqlalchemy/oop_approach.py new file mode 100644 index 00000000..a4bab7f1 --- /dev/null +++ b/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"" + + +def main() -> None: + Base.metadata.create_all(db) + user = User(username="Arjan", email="Arjan@arjancodes.com") + + with Session() as session: + session.add(user) + session.commit() + print(session.query(User).all()) + + +if __name__ == "__main__": + main() diff --git a/2024/sqlalchemy/pyproject.toml b/2024/sqlalchemy/pyproject.toml new file mode 100644 index 00000000..11a6551f --- /dev/null +++ b/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" diff --git a/2024/sqlalchemy/relationship.py b/2024/sqlalchemy/relationship.py new file mode 100644 index 00000000..13b4027c --- /dev/null +++ b/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"" + + +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"" + + +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"" + + +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) + 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() diff --git a/2024/sqlalchemy/simple_approach.py b/2024/sqlalchemy/simple_approach.py new file mode 100644 index 00000000..1b9b5410 --- /dev/null +++ b/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", "Arjan@arjancodes.com") + print(select_user("Arjan")) + connection.close() + + +if __name__ == "__main__": + main()