This commit is contained in:
JoaoBastos023 2024-11-12 10:27:37 +00:00
commit 89de230367
12 changed files with 374 additions and 0 deletions

89
.gitignore vendored Normal file
View File

@ -0,0 +1,89 @@
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
env/
venv/
.venv/
ENV/
env.bak/
venv.bak/
*.egg
*.egg-info/
dist/
build/
*.whl
# Flask specific
instance/
.webassets-cache
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
.python-version
# celery beat schedule file
celerybeat-schedule
# editors
*~
.vscode/
.idea/
# database
*.db

View File

@ -0,0 +1,11 @@
import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
script_dir = os.path.dirname(os.path.abspath(__file__))
DATABASE_URL = f"sqlite:///{os.path.join(script_dir, 'database.db')}"
engine = create_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

View File

@ -0,0 +1,13 @@
import os
from database import Base, engine
from models import Organization, User, File
def setup_db():
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "database.db")
if os.path.exists(db_path):
os.remove(db_path)
Base.metadata.create_all(bind=engine)
if __name__ == "__main__":
setup_db()

View File

@ -0,0 +1,3 @@
from .user import *
from .org import *
from .file import *

View File

@ -0,0 +1,19 @@
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
from database import Base
from dataclasses import dataclass
@dataclass
class File(Base):
__tablename__ = 'files'
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
file_handle: Mapped[str] = mapped_column(String, nullable=False)
document_handle: Mapped[str] = mapped_column(String, nullable=False)
name: Mapped[str] = mapped_column(String, nullable=False)
created_at: Mapped[int] = mapped_column(Integer, nullable=False)
org_id: Mapped[int] = mapped_column(Integer, ForeignKey('organizations.id'), nullable=False)
creator_id: Mapped[int] = mapped_column(Integer, ForeignKey('users.id'), nullable=False)
org: Mapped['Organization'] = relationship('Organization', back_populates='files')
creator: Mapped['User'] = relationship('User', back_populates='files')

View File

@ -0,0 +1,14 @@
from sqlalchemy import Integer, String, ForeignKey
from sqlalchemy.orm import relationship, Mapped, mapped_column
from database import Base
from dataclasses import dataclass
@dataclass
class Organization(Base):
__tablename__ = 'organizations'
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
name: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
owner_id: Mapped[int] = mapped_column(Integer, ForeignKey('users.id'), nullable=False)
owner: Mapped['User'] = relationship('User', back_populates='orgs')
files: Mapped[list['File']] = relationship('File', back_populates='org')

View File

@ -0,0 +1,16 @@
from sqlalchemy import Integer, String
from sqlalchemy.orm import relationship, Mapped, mapped_column
from database import Base
from dataclasses import dataclass
@dataclass
class User(Base):
__tablename__ = 'users'
id: Mapped[int] = mapped_column(Integer, primary_key=True, index=True)
username: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
full_name: Mapped[str] = mapped_column(String, nullable=False)
email: Mapped[str] = mapped_column(String, unique=True, index=True, nullable=False)
public_key: Mapped[str] = mapped_column(String, nullable=False)
orgs: Mapped[list['Organization']] = relationship('Organization', back_populates='owner')
files: Mapped[list['File']] = relationship('File', back_populates='creator')

View File

@ -0,0 +1,3 @@
flask
sqlalchemy
pytest

View File

@ -0,0 +1,2 @@
from .orgs import OrganizationService
from .users import UserService

View File

@ -0,0 +1,36 @@
from sqlalchemy.orm import Session
from models import Organization
class OrganizationService:
def __init__(self, db: Session):
self.db = db
def create_organization(self, name: str, username: str, full_name: str, email: str, public_key: str) -> Organization:
from services import UserService
user = UserService(self.db).get_user_by_username(username)
if not user:
user = UserService(self.db).create_user(username, full_name, email, public_key)
organization = Organization(
name=name,
owner_id=user.id,
owner=user
)
self.db.add(organization)
self.db.commit()
self.db.refresh(organization)
UserService(self.db).add_org_to_user(user, organization)
return organization
def get_organization(self, org_id: int) -> Organization | None:
return self.db.query(Organization).filter(Organization.id == org_id).first()
def get_organization_by_name(self, name: str) -> Organization | None:
return self.db.query(Organization).filter(Organization.name == name).first()
def list_organizations(self):
return self.db.query(Organization).all()

View File

@ -0,0 +1,32 @@
from sqlalchemy.orm import Session
from models import User, Organization
class UserService:
def __init__(self, db: Session):
self.db = db
def create_user(self, username: str, full_name: str, email: str, public_key: str, org: Organization = None) -> User:
user = User(
username=username,
full_name=full_name,
email=email,
public_key=public_key,
orgs=[org] if org else []
)
self.db.add(user)
self.db.commit()
self.db.refresh(user)
return user
def get_user(self, user_id: int) -> User | None:
return self.db.query(User).filter(User.id == user_id).first()
def get_user_by_username(self, username: str) -> User | None:
return self.db.query(User).filter(User.username == username).first()
def add_org_to_user(self, user: User, org: Organization) -> User:
user.orgs.append(org)
self.db.commit()
self.db.refresh(user)
return user

View File

@ -0,0 +1,136 @@
from pprint import PrettyPrinter
import database.setup_db
from database import SessionLocal
from services import OrganizationService, UserService
database.setup_db.setup_db()
db = SessionLocal()
org_service = OrganizationService(db)
user_service = UserService(db)
def test_create_organization():
org = org_service.create_organization(
name="Test Org",
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
assert org.name == "Test Org"
assert org.owner.username == "testuser"
assert org.owner.full_name == "Test User"
assert org.owner.email == "test@mail.com"
assert org.owner.public_key == "abc123"
assert org.owner.orgs[0].name == "Test Org"
db.delete(org.owner)
db.delete(org)
db.commit()
def test_create_user():
user = user_service.create_user(
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
assert user.username == "testuser"
assert user.full_name == "Test User"
assert user.email == "test@mail.com"
assert user.public_key == "abc123"
db.delete(user)
db.commit()
def test_get_organization():
org = org_service.create_organization(
name="Test Org",
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
org2 = org_service.get_organization(org.id)
org3 = org_service.get_organization_by_name("Test Org")
assert org2.name == "Test Org"
assert org3.name == "Test Org"
assert org2.owner.username == "testuser"
assert org3.owner.username == "testuser"
db.delete(org.owner)
db.delete(org)
db.commit()
def test_list_organizations():
org = org_service.create_organization(
name="Test Org",
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
org2 = org_service.create_organization(
name="Test Org2",
username="testuser2",
full_name="Test User2",
email="mail2@test.com",
public_key="def456"
)
orgs = org_service.list_organizations()
assert len(orgs) == 2
assert orgs[0].name == "Test Org"
assert orgs[1].name == "Test Org2"
assert orgs == [org, org2]
db.delete(org.owner)
db.delete(org)
db.delete(org2.owner)
db.delete(org2)
db.commit()
def test_get_user():
user = user_service.create_user(
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
user2 = user_service.get_user(user.id)
user3 = user_service.get_user_by_username("testuser")
assert user2.username == "testuser"
assert user3.username == "testuser"
assert user2.full_name == "Test User"
assert user3.full_name == "Test User"
assert user2.email == "test@mail.com"
assert user3.email == "test@mail.com"
assert user2.public_key == "abc123"
assert user3.public_key == "abc123"
org = org_service.create_organization(
name="Test Org",
username="testuser",
full_name="Test User",
email="test@mail.com",
public_key="abc123"
)
user4 = user_service.get_user(user.id)
assert user4.orgs[0].name == "Test Org"
db.delete(org)
db.delete(user)
db.commit()