From 7b8503ca5c27dfa681f8ca4d919368528a1d01bb Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Tue, 13 Feb 2024 20:39:33 +0000 Subject: [PATCH 1/8] Add files via upload --- 2024/sqlalchemy_example/oop_approach.py | 28 ++++++++ 2024/sqlalchemy_example/relationship.py | 79 ++++++++++++++++++++++ 2024/sqlalchemy_example/simple_approach.py | 32 +++++++++ 3 files changed, 139 insertions(+) create mode 100644 2024/sqlalchemy_example/oop_approach.py create mode 100644 2024/sqlalchemy_example/relationship.py create mode 100644 2024/sqlalchemy_example/simple_approach.py diff --git a/2024/sqlalchemy_example/oop_approach.py b/2024/sqlalchemy_example/oop_approach.py new file mode 100644 index 00000000..93f4771c --- /dev/null +++ b/2024/sqlalchemy_example/oop_approach.py @@ -0,0 +1,28 @@ +import sqlalchemy as sa +from sqlalchemy.orm import sessionmaker, declarative_base + +db = sa.create_engine('sqlite:///:memory:') +Session = sessionmaker(bind=db) +Base = declarative_base() + +class User(Base): + __tablename__ = 'users' + + id: int = sa.Column(sa.Integer, primary_key=True) + username:str = sa.Column(sa.String) + email: str = sa.Column(sa.String) + + def __repr__(self): + return f'' + +def main(): + 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() \ No newline at end of file diff --git a/2024/sqlalchemy_example/relationship.py b/2024/sqlalchemy_example/relationship.py new file mode 100644 index 00000000..fe3651bb --- /dev/null +++ b/2024/sqlalchemy_example/relationship.py @@ -0,0 +1,79 @@ +import hashlib +import typing +import sqlalchemy as sa +from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Mapped + +db = sa.create_engine('sqlite:///:memory:') +Session = sessionmaker(bind=db) +Base = declarative_base() + +class User(Base): + __tablename__ = 'users' + id:int = sa.Column(sa.Integer, 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, email, password): + self.auth = UserAuth(username=username, email=email) + self.auth.set_password(password) + + def __repr__(self): + 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, email): + self.username = username + self.email = email + + def set_password(self, password): + self.password_hash = hashlib.sha256(password.encode('utf-8')).hexdigest() + + def check_password(self, password): + return self.password_hash == hashlib.sha256(password.encode('utf-8')).hexdigest() + + def __repr__(self): + 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): + return f'' + + +def main(): + 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() \ No newline at end of file diff --git a/2024/sqlalchemy_example/simple_approach.py b/2024/sqlalchemy_example/simple_approach.py new file mode 100644 index 00000000..83e2d945 --- /dev/null +++ b/2024/sqlalchemy_example/simple_approach.py @@ -0,0 +1,32 @@ +import sqlalchemy as sa +import atexit + +engine = sa.create_engine('sqlite:///:memory:') +connection = engine.connect() + +atexit.register(connection.close) +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, email): + query = user_table.insert().values(username=username, email=email) + connection.execute(query) + +def select_user(username): + query = user_table.select().where(user_table.c.username == username) + result = connection.execute(query) + return result.fetchone() + +def main(): + metadata.create_all(engine) + insert_user("Arjan", "Arjan@arjancodes.com") + print(select_user("Arjan")) + +if __name__ == '__main__': + main() From b1b02d1b77fe4c5d09e8705e31759740afc8b7d5 Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Tue, 13 Feb 2024 20:43:30 +0000 Subject: [PATCH 2/8] Add files via upload --- 2024/sqlalchemy_example/oop_approach.py | 23 ++++---- 2024/sqlalchemy_example/relationship.py | 66 ++++++++++++---------- 2024/sqlalchemy_example/simple_approach.py | 11 ++-- 3 files changed, 55 insertions(+), 45 deletions(-) diff --git a/2024/sqlalchemy_example/oop_approach.py b/2024/sqlalchemy_example/oop_approach.py index 93f4771c..c4db2ddb 100644 --- a/2024/sqlalchemy_example/oop_approach.py +++ b/2024/sqlalchemy_example/oop_approach.py @@ -2,27 +2,30 @@ from sqlalchemy.orm import sessionmaker, declarative_base db = sa.create_engine('sqlite:///:memory:') -Session = sessionmaker(bind=db) +Session = sessionmaker(bind = db) Base = declarative_base() + class User(Base): __tablename__ = 'users' - - id: int = sa.Column(sa.Integer, primary_key=True) - username:str = sa.Column(sa.String) + + id: int = sa.Column(sa.Integer, primary_key = True) + username: str = sa.Column(sa.String) email: str = sa.Column(sa.String) - + def __repr__(self): return f'' - + + def main(): 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) session.commit() print(session.query(User).all()) - + + if __name__ == '__main__': - main() \ No newline at end of file + main() diff --git a/2024/sqlalchemy_example/relationship.py b/2024/sqlalchemy_example/relationship.py index fe3651bb..82fa196b 100644 --- a/2024/sqlalchemy_example/relationship.py +++ b/2024/sqlalchemy_example/relationship.py @@ -4,76 +4,80 @@ from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Mapped db = sa.create_engine('sqlite:///:memory:') -Session = sessionmaker(bind=db) +Session = sessionmaker(bind = db) Base = declarative_base() + class User(Base): __tablename__ = 'users' - id:int = sa.Column(sa.Integer, primary_key=True) - auth: Mapped["UserAuth"] = relationship('UserAuth', uselist=False, back_populates='user') - posts: Mapped[typing.List["UserPost"]] = relationship('UserPost', back_populates= 'user') - + id: int = sa.Column(sa.Integer, 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, email, password): - self.auth = UserAuth(username=username, email=email) + super().__init__() + self.auth = UserAuth(username = username, email = email) self.auth.set_password(password) - + def __repr__(self): 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') - + + 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, email): self.username = username self.email = email - + def set_password(self, password): self.password_hash = hashlib.sha256(password.encode('utf-8')).hexdigest() - + def check_password(self, password): return self.password_hash == hashlib.sha256(password.encode('utf-8')).hexdigest() - + def __repr__(self): 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') - + 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): return f'' - + def main(): 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() \ No newline at end of file + main() diff --git a/2024/sqlalchemy_example/simple_approach.py b/2024/sqlalchemy_example/simple_approach.py index 83e2d945..2974cb2a 100644 --- a/2024/sqlalchemy_example/simple_approach.py +++ b/2024/sqlalchemy_example/simple_approach.py @@ -8,25 +8,28 @@ metadata = sa.MetaData() user_table = sa.Table('user', metadata, - sa.Column('id', sa.Integer, primary_key=True), + sa.Column('id', sa.Integer, primary_key = True), sa.Column('username', sa.String), sa.Column('email', sa.String), ) def insert_user(username, email): - query = user_table.insert().values(username=username, email=email) + query = user_table.insert().values(username = username, email = email) connection.execute(query) - + + def select_user(username): query = user_table.select().where(user_table.c.username == username) result = connection.execute(query) return result.fetchone() + def main(): metadata.create_all(engine) insert_user("Arjan", "Arjan@arjancodes.com") print(select_user("Arjan")) - + + if __name__ == '__main__': main() From 4a07d108e71d85486f7db5fe4b49cb4502ea3a3c Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Fri, 1 Mar 2024 12:03:15 +0000 Subject: [PATCH 3/8] Delete 2024/sqlalchemy_example directory --- 2024/sqlalchemy_example/oop_approach.py | 31 -------- 2024/sqlalchemy_example/relationship.py | 83 ---------------------- 2024/sqlalchemy_example/simple_approach.py | 35 --------- 3 files changed, 149 deletions(-) delete mode 100644 2024/sqlalchemy_example/oop_approach.py delete mode 100644 2024/sqlalchemy_example/relationship.py delete mode 100644 2024/sqlalchemy_example/simple_approach.py diff --git a/2024/sqlalchemy_example/oop_approach.py b/2024/sqlalchemy_example/oop_approach.py deleted file mode 100644 index c4db2ddb..00000000 --- a/2024/sqlalchemy_example/oop_approach.py +++ /dev/null @@ -1,31 +0,0 @@ -import sqlalchemy as sa -from sqlalchemy.orm import sessionmaker, declarative_base - -db = sa.create_engine('sqlite:///:memory:') -Session = sessionmaker(bind = db) -Base = declarative_base() - - -class User(Base): - __tablename__ = 'users' - - id: int = sa.Column(sa.Integer, primary_key = True) - username: str = sa.Column(sa.String) - email: str = sa.Column(sa.String) - - def __repr__(self): - return f'' - - -def main(): - 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_example/relationship.py b/2024/sqlalchemy_example/relationship.py deleted file mode 100644 index 82fa196b..00000000 --- a/2024/sqlalchemy_example/relationship.py +++ /dev/null @@ -1,83 +0,0 @@ -import hashlib -import typing -import sqlalchemy as sa -from sqlalchemy.orm import declarative_base, relationship, sessionmaker, Mapped - -db = sa.create_engine('sqlite:///:memory:') -Session = sessionmaker(bind = db) -Base = declarative_base() - - -class User(Base): - __tablename__ = 'users' - id: int = sa.Column(sa.Integer, 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, email, password): - super().__init__() - self.auth = UserAuth(username = username, email = email) - self.auth.set_password(password) - - def __repr__(self): - 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, email): - self.username = username - self.email = email - - def set_password(self, password): - self.password_hash = hashlib.sha256(password.encode('utf-8')).hexdigest() - - def check_password(self, password): - return self.password_hash == hashlib.sha256(password.encode('utf-8')).hexdigest() - - def __repr__(self): - 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): - return f'' - - -def main(): - 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_example/simple_approach.py b/2024/sqlalchemy_example/simple_approach.py deleted file mode 100644 index 2974cb2a..00000000 --- a/2024/sqlalchemy_example/simple_approach.py +++ /dev/null @@ -1,35 +0,0 @@ -import sqlalchemy as sa -import atexit - -engine = sa.create_engine('sqlite:///:memory:') -connection = engine.connect() - -atexit.register(connection.close) -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, email): - query = user_table.insert().values(username = username, email = email) - connection.execute(query) - - -def select_user(username): - query = user_table.select().where(user_table.c.username == username) - result = connection.execute(query) - return result.fetchone() - - -def main(): - metadata.create_all(engine) - insert_user("Arjan", "Arjan@arjancodes.com") - print(select_user("Arjan")) - - -if __name__ == '__main__': - main() From d487406c400dbb29d78fed6d8c482d026c3f10a0 Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Fri, 1 Mar 2024 12:04:08 +0000 Subject: [PATCH 4/8] Add files via upload --- 2024/sqlalchemy/oop_approach.py | 31 +++++++++++ 2024/sqlalchemy/relationship.py | 83 ++++++++++++++++++++++++++++++ 2024/sqlalchemy/simple_approach.py | 37 +++++++++++++ 3 files changed, 151 insertions(+) create mode 100644 2024/sqlalchemy/oop_approach.py create mode 100644 2024/sqlalchemy/relationship.py create mode 100644 2024/sqlalchemy/simple_approach.py diff --git a/2024/sqlalchemy/oop_approach.py b/2024/sqlalchemy/oop_approach.py new file mode 100644 index 00000000..56197c0c --- /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: str = sa.Column(sa.String) + email: str = sa.Column(sa.String) + + 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/relationship.py b/2024/sqlalchemy/relationship.py new file mode 100644 index 00000000..7b20b28f --- /dev/null +++ b/2024/sqlalchemy/relationship.py @@ -0,0 +1,83 @@ +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..ebf351cd --- /dev/null +++ b/2024/sqlalchemy/simple_approach.py @@ -0,0 +1,37 @@ +import atexit + +import sqlalchemy as sa + +engine = sa.create_engine('sqlite:///:memory:') +connection = engine.connect() + +atexit.register(connection.close) +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, email): + query = user_table.insert().values(username = username, email = email) + connection.execute(query) + + +def select_user(username): + query = user_table.select().where(user_table.c.username == username) + result = connection.execute(query) + return result.fetchone() + + +def main(): + metadata.create_all(engine) + insert_user("Arjan", "Arjan@arjancodes.com") + print(select_user("Arjan")) + + +if __name__ == '__main__': + main() From 05c7d36673a78270eb77061f2ce13c77d71b09a7 Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Fri, 1 Mar 2024 12:10:45 +0000 Subject: [PATCH 5/8] Add files via upload --- 2024/sqlalchemy/pyproject.toml | 13 +++++++++++++ 1 file changed, 13 insertions(+) create mode 100644 2024/sqlalchemy/pyproject.toml 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" From 89aa32648b8253e27acc6ba4bc2f7ae5b81c5a24 Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Fri, 1 Mar 2024 12:12:58 +0000 Subject: [PATCH 6/8] Update simple_approach.py --- 2024/sqlalchemy/simple_approach.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/2024/sqlalchemy/simple_approach.py b/2024/sqlalchemy/simple_approach.py index ebf351cd..488cbbdd 100644 --- a/2024/sqlalchemy/simple_approach.py +++ b/2024/sqlalchemy/simple_approach.py @@ -1,11 +1,8 @@ -import atexit - import sqlalchemy as sa engine = sa.create_engine('sqlite:///:memory:') connection = engine.connect() -atexit.register(connection.close) metadata = sa.MetaData() user_table = sa.Table( @@ -31,6 +28,7 @@ def main(): metadata.create_all(engine) insert_user("Arjan", "Arjan@arjancodes.com") print(select_user("Arjan")) + connection.close() if __name__ == '__main__': From 5c17b294ddbcfc8d9f0cd9fe30b43acbcf1ec2dd Mon Sep 17 00:00:00 2001 From: MissArticulatePython <39281408+AlyceOsbourne@users.noreply.github.com> Date: Fri, 1 Mar 2024 12:23:16 +0000 Subject: [PATCH 7/8] Update simple_approach.py --- 2024/sqlalchemy/simple_approach.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/2024/sqlalchemy/simple_approach.py b/2024/sqlalchemy/simple_approach.py index 488cbbdd..5c835616 100644 --- a/2024/sqlalchemy/simple_approach.py +++ b/2024/sqlalchemy/simple_approach.py @@ -13,18 +13,18 @@ ) -def insert_user(username, 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): +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(): +def main() -> None: metadata.create_all(engine) insert_user("Arjan", "Arjan@arjancodes.com") print(select_user("Arjan")) From 2702fa05a1b939e6cfd50c287f52e76157e3cbfe Mon Sep 17 00:00:00 2001 From: Arjan Egges Date: Tue, 5 Mar 2024 14:29:44 +0100 Subject: [PATCH 8/8] 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()