finish for init core

fuware/app.py

@@ -0,0 +1,60 @@
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import JSONResponse
+from fastapi.middleware.cors import CORSMiddleware
+from fastapi.middleware.gzip import GZipMiddleware
+
+from fuware.core.config import get_app_settings
+from fuware import __version__
+from fuware.routes import router
+import uvicorn
+
+settings = get_app_settings()
+
+description = f"""
+fuware is a web application for managing your hours items and tracking them.
+"""
+
+# event.listen(models.User.__table__, 'after_create', initialize_table)
+
+app = FastAPI(
+  title="Fuware",
+  description=description,
+  version=__version__,
+  docs_url=settings.DOCS_URL,
+  redoc_url=settings.REDOC_URL
+)
+
+app.add_middleware(GZipMiddleware, minimum_size=1000)
+
+if not settings.PRODUCTION:
+  allowed_origins = ["http://localhost:3000"]
+
+  app.add_middleware(
+    CORSMiddleware,
+    allow_origins=allowed_origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+  )
+
+@app.exception_handler(HTTPException)
+async def unicorn_exception_handler(request: Request, exc: HTTPException):
+  return JSONResponse(
+    status_code=exc.status_code,
+    content={"status": exc.status_code, "data": exc.detail},
+  )
+
+def api_routers():
+  app.include_router(router)
+
+
+api_routers()
+
+# app.include_router(authR.authRouter)
+# app.include_router(userR.userRouter)
+
+def main():
+  uvicorn.run("app:app", host="0.0.0.0", port=settings.API_PORT, reload=True, workers=1, forwarded_allow_ips="*")
+
+if __name__ == "__main__":
+  main()

fuware/const.py

@@ -1,15 +0,0 @@
-import os
-from dotenv import load_dotenv
-
-load_dotenv()
-
-SERCET_KEY = b"oWNhXlfo666JlMHk6UHYxeNB6z_CA2MisDDZJe4N0yc="
-COOKIE_KEY = os.getenv('VITE_LOGIN_KEY') or '7fo24CMyIc'
-# URL_DATABASE = "postgresql://{0}:{1}@{2}:{3}/{4}".format(
-#   os.getenv('LOL_DB_USER'),
-#   os.getenv('LOL_DB_PASSWORD'),
-#   os.getenv('LOL_DB_HOST'),
-#   os.getenv('LOL_DB_PORT'),
-#   os.getenv('LOL_DB_NAME'),
-# )
-URL_DATABASE = "sqlite:///./test.db"

fuware/core/config.py

@@ -4,8 +4,7 @@ from pathlib import Path
 
 from dotenv import load_dotenv
 
-
-from .settings import AppDirectories, AppSettings
+from fuware.core.settings.settings import AppSettings, app_settings_constructor
 
 CWD = Path(__file__).parent
 BASE_DIR = CWD.parent.parent
@@ -27,14 +26,6 @@ def determine_data_dir() -> Path:
 
     return BASE_DIR.joinpath("dev", "data")
 
-@lru_cache
-def get_app_dirs() -> AppDirectories:
-    return AppDirectories(determine_data_dir())
-
-
 @lru_cache
 def get_app_settings() -> AppSettings:
     return app_settings_constructor(env_file=ENV, production=PRODUCTION, data_dir=determine_data_dir())
-
-
-print(get_app_settings())

fuware/core/dependencies/__init__.py

@@ -0,0 +1 @@
+from .dependencies import *

fuware/core/dependencies/dependencies.py

@@ -0,0 +1,20 @@
+
+from fastapi import Depends, HTTPException, Request
+from sqlalchemy.orm import Session
+from fuware.core.config import get_app_settings
+from fuware.db.db_setup import generate_session
+
+settings = get_app_settings()
+
+async def get_auth_user(request: Request, db: Session = Depends(generate_session)):
+  """verify that user has a valid session"""
+  session_id = request.cookies.get(settings.COOKIE_KEY)
+  if not session_id:
+    raise HTTPException(status_code=401, detail="Unauthorized")
+  # decrypt_user = decryptString(session_id).split(',')
+  # db_user = get_user_by_username(db, decrypt_user[0])
+  # if not db_user:
+  #   raise HTTPException(status_code=403)
+  # if not verify_password(decrypt_user[1], db_user.password):
+  #   raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
+  return True

fuware/core/logger/config.py

@@ -0,0 +1,67 @@
+import json
+import logging
+import pathlib
+import typing
+from logging import config as logging_config
+
+__dir = pathlib.Path(__file__).parent
+__conf: dict[str, str] | None = None
+
+
+def _load_config(path: pathlib.Path, substitutions: dict[str, str] | None = None) -> dict[str, typing.Any]:
+  with open(path) as file:
+    if substitutions:
+      contents = file.read()
+      for key, value in substitutions.items():
+        # Replaces the key matches
+        #
+        # Example:
+        #   {"key": "value"}
+        #   "/path/to/${key}/file" -> "/path/to/value/file"
+        contents = contents.replace(f"${{{key}}}", value)
+
+      json_data = json.loads(contents)
+
+    else:
+      json_data = json.load(file)
+
+  return json_data
+
+
+def log_config() -> dict[str, str]:
+  if __conf is None:
+    raise ValueError("logger not configured, must call configured_logger first")
+
+  return __conf
+
+
+def configured_logger(
+  *,
+  mode: str,
+  config_override: pathlib.Path | None = None,
+  substitutions: dict[str, str] | None = None,
+) -> logging.Logger:
+  """
+  Configure the logger based on the mode and return the root logger
+
+  Args:
+      mode (str): The mode to configure the logger for (production, development, testing)
+      config_override (pathlib.Path, optional): A path to a custom logging config. Defaults to None.
+      substitutions (dict[str, str], optional): A dictionary of substitutions to apply to the logging config.
+  """
+  global __conf
+
+  if config_override:
+    __conf = _load_config(config_override, substitutions)
+  else:
+    if mode == "production":
+      __conf = _load_config(__dir / "logconf.prod.json", substitutions)
+    elif mode == "development":
+      __conf = _load_config(__dir / "logconf.dev.json", substitutions)
+    elif mode == "testing":
+      __conf = _load_config(__dir / "logconf.test.json", substitutions)
+    else:
+      raise ValueError(f"Invalid mode: {mode}")
+
+  logging_config.dictConfig(config=__conf)
+  return logging.getLogger()

fuware/core/security/hasher.py

@@ -0,0 +1,34 @@
+from functools import lru_cache
+from typing import Protocol
+import bcrypt
+
+from fuware.core.config import get_app_settings
+
+
+class Hasher(Protocol):
+  def hash(self, password: str) -> str: ...
+
+  def verify(self, password: str, hashed: str) -> bool: ...
+
+class FakeHasher:
+  def hash(self, password: str) -> str:
+    return password
+
+  def verify(self, password: str, hashed: str) -> bool:
+    return password == hashed
+
+class BcryptHasher:
+  def hash(self, password: str) -> str:
+    return bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+
+  def verify(self, password: str, hashed: str) -> bool:
+    return bcrypt.checkpw(password.encode('utf-8'), hashed.encode('utf-8'))
+
+@lru_cache(maxsize=1)
+def get_hasher() -> Hasher:
+    settings = get_app_settings()
+
+    if settings.TESTING:
+        return FakeHasher()
+
+    return BcryptHasher()

fuware/core/settings/db_providers.py

@@ -17,7 +17,7 @@ class SQLiteProvider(AbstractDBProvider, BaseModel):
 
   @property
   def db_path(self):
-    return self.data_dir / f"{self.prefix}mealie.db"
+    return self.data_dir / f"{self.prefix}fuware.db"
 
   @property
   def db_url(self) -> str:

fuware/core/settings/settings.py

@@ -1,26 +1,23 @@
-import secrets
 from pathlib import Path
 from fuware.core.settings.db_providers import AbstractDBProvider, SQLiteProvider
 from pydantic_settings import BaseSettings # type: ignore
 
 
-def determine_secrets(data_dir: Path, production: bool) -> str:
+def determine_secrets(production: bool) -> str:
   if not production:
     return "shh-secret-test-key"
 
-  secrets_file = data_dir.joinpath(".secret")
-  if secrets_file.is_file():
-    with open(secrets_file) as f:
-      return f.read()
-  else:
-    data_dir.mkdir(parents=True, exist_ok=True)
-    with open(secrets_file, "w") as f:
-        new_secret = secrets.token_hex(32)
-        f.write(new_secret)
-    return new_secret
+  return "oWNhXlfo666JlMHk6UHYxeNB6z_CA2MisDDZJe4N0yc="
+
+def determine_cookie(production: bool) -> str:
+  if not production:
+    return "logcook"
+
+  return "7fo24CMyIc"
 
 class AppSettings(BaseSettings):
   PRODUCTION: bool
+  TESTING: bool
   BASE_URL: str = "http://localhost:8080"
   """trailing slashes are trimmed (ex. `http://localhost:8080/` becomes ``http://localhost:8080`)"""
 
@@ -32,6 +29,9 @@ class AppSettings(BaseSettings):
 
   ALLOW_SIGNUP: bool = False
 
+  SECRET: str
+  COOKIE_KEY: str
+
   @property
   def DOCS_URL(self) -> str | None:
     return "/docs" if self.API_DOCS else None
@@ -43,7 +43,6 @@ class AppSettings(BaseSettings):
   # ===============================================
   # Database Configuration
 
-  DB_ENGINE: str = "sqlite"  # Options: 'sqlite', 'postgres'
   DB_PROVIDER: AbstractDBProvider | None = None
 
   @property
@@ -63,7 +62,7 @@ def app_settings_constructor(data_dir: Path, production: bool, env_file: Path, e
   app_settings = AppSettings(
     _env_file=env_file,  # type: ignore
     _env_file_encoding=env_encoding,  # type: ignore
-    **{"SECRET": determine_secrets(data_dir, production)},
+    **{"SECRET": determine_secrets(production), 'COOKIE_KEY': determine_cookie(production)},
   )
 
   app_settings.DB_PROVIDER = SQLiteProvider(data_dir=data_dir)

fuware/core/settings/static.py

@@ -1,22 +0,0 @@
-import os
-from dotenv import load_dotenv
-
-from pathlib import Path
-from fuware import __version__
-
-load_dotenv()
-
-APP_VERSION = __version__
-CWD = Path(__file__).parent
-BASE_DIR = CWD.parent.parent.parent
-
-SERCET_KEY = b"oWNhXlfo666JlMHk6UHYxeNB6z_CA2MisDDZJe4N0yc="
-COOKIE_KEY = os.getenv('VITE_LOGIN_KEY') or '7fo24CMyIc'
-# URL_DATABASE = "postgresql://{0}:{1}@{2}:{3}/{4}".format(
-#   os.getenv('LOL_DB_USER'),
-#   os.getenv('LOL_DB_PASSWORD'),
-#   os.getenv('LOL_DB_HOST'),
-#   os.getenv('LOL_DB_PORT'),
-#   os.getenv('LOL_DB_NAME'),
-# )
-URL_DATABASE = "sqlite:///./test.db"

fuware/db/controller/user.py

@@ -1,20 +0,0 @@
-from sqlalchemy.orm import Session
-from db.models import User
-from ultis import get_password_hash
-import schemas
-
-def get_user(db: Session, user_id: str):
-  return db.query(User).filter(User.id == user_id).first()
-
-def get_user_by_username(db: Session, usn: str):
-  return db.query(User).filter(User.username == usn).first()
-
-def get_users(db: Session, skip: int = 0, limit: int = 100):
-  return db.query(User).offset(skip).limit(limit).all()
-
-def create_user(db: Session, user: schemas.UserCreate):
-  db_user = User(username=user.username, password=get_password_hash(user.password), name=user.name)
-  db.add(db_user)
-  db.commit()
-  db.refresh(db_user)
-  return db_user

fuware/db/db_setup.py

@@ -1,17 +1,37 @@
-from sqlalchemy import create_engine
-from sqlalchemy.orm import sessionmaker
-from sqlalchemy.ext.declarative import declarative_base
-from const import URL_DATABASE
+from collections.abc import Generator
+from sqlalchemy.orm.session import Session
+from sqlalchemy import create_engine, event, Engine, text
+from sqlalchemy.orm import scoped_session, sessionmaker
 
-engine = create_engine(URL_DATABASE)
+from fuware.core.config import get_app_settings
 
-SessionLocal = sessionmaker(autocommit=False ,autoflush=False, bind=engine)
+settings = get_app_settings()
 
-Base = declarative_base()
+def sql_global_init(db_url: str):
+  connect_args = {"check_same_thread": False}
 
-def get_db():
+  engine = create_engine(db_url, echo=True, connect_args=connect_args, pool_pre_ping=True, future=True)
+
+  SessionLocal = scoped_session(sessionmaker(autocommit=False, autoflush=False, bind=engine, future=True))
+
+  return SessionLocal, engine
+
+SessionLocal, engine = sql_global_init(settings.DB_URL)  # type: ignore
+
+@event.listens_for(Engine, "connect")
+def set_sqlite_pragma(dbapi_connection, connection_record):
+  cursor = dbapi_connection.cursor()
+  cursor.execute("PRAGMA foreign_keys=ON")
+  cursor.close()
+
+# with engine.connect() as connection:
+#   result = connection.execute(text('select "Hello"'))
+
+#   print(result.all())
+
+def generate_session() -> Generator[Session, None, None]:
   db = SessionLocal()
   try:
-    yield db
+      yield db
   finally:
-    db.close()
+      db.close()

fuware/db/init_db.py

@@ -0,0 +1,5 @@
+from db_setup import engine
+from models._model_base import Model
+from models.users import *
+
+Model.metadata.create_all(bind=engine)

fuware/db/models/__init__.py

@@ -1 +1 @@
-from .user import *
+from .users import *

fuware/db/models/_model_base.py

@@ -0,0 +1,20 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer
+from sqlalchemy.orm import declarative_base, Mapped, mapped_column
+from text_unidecode import unidecode
+
+from fuware.db.db_setup import SessionLocal
+
+Model = declarative_base()
+Model.query = SessionLocal.query_property()
+
+class SqlAlchemyBase(Model):
+  __abstract__ = True
+
+  created_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow(), index=True)
+  update_at: Mapped[datetime | None] = mapped_column(DateTime, default=datetime.utcnow(), onupdate=datetime.utcnow())
+
+  @classmethod
+  def normalize(cls, val: str) -> str:
+    return unidecode(val).lower().strip()

fuware/db/models/mixins.py

@@ -1,8 +0,0 @@
-from datetime import datetime
-from sqlalchemy import Column, DateTime
-from sqlalchemy.orm import declarative_mixin
-
-@declarative_mixin
-class Timestamp:
-  created_at = Column(DateTime, default=datetime.utcnow, nullable=False)
-  updated_at = Column(DateTime, default=datetime.utcnow, nullable=False)

fuware/db/models/user.py

@@ -1,15 +0,0 @@
-from db import Base
-from sqlalchemy import Boolean, Column, String
-from .mixins import Timestamp
-from sqlalchemy.dialects.postgresql import UUID
-import uuid
-
-class User(Base, Timestamp):
-  __tablename__ = 'users'
-
-  id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, index=True)
-  username = Column(String(100), unique=True, index=True, nullable=False)
-  password = Column(String, index=True, nullable=False)
-  name = Column(String, index=True, nullable=True)
-  is_admin = Column(Boolean, default=False)
-  is_lock = Column(Boolean, default=False)

fuware/db/models/users/__init__.py

@@ -0,0 +1 @@
+from .users import *

fuware/db/models/users/users.py

@@ -0,0 +1,20 @@
+import uuid
+from sqlalchemy import Boolean, Column, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from sqlalchemy.dialects.postgresql import UUID
+
+from .._model_base import SqlAlchemyBase
+
+class User(SqlAlchemyBase):
+  __tablename__ = 'users'
+
+  id: Mapped[UUID] = mapped_column(UUID, primary_key=True, default=uuid.uuid4, index=True)
+  username: Mapped[str | None] = mapped_column(String, unique=True, index=True, nullable=False)
+  password: Mapped[str | None] = mapped_column(String, index=True, nullable=False)
+  name: Mapped[str | None] = mapped_column(String, index=True, nullable=True)
+  is_admin: Mapped[bool | None] = mapped_column(Boolean, default=False)
+  is_lock: Mapped[bool | None] = mapped_column(Boolean, default=False)
+
+  def __repr__(self):
+    return f"{self.__class__.__name__}, name: {self.name}, username: {self.username}"

fuware/main.py

@@ -1,34 +1,10 @@
-from fastapi import FastAPI, Request, HTTPException
-from fastapi.responses import JSONResponse
-from routes import authR, userR
-# from db import engine, models
-# from sqlalchemy import event
-# from db.seeds import initialize_table
+
 import uvicorn
+from fuware.app import settings
 
-# event.listen(models.User.__table__, 'after_create', initialize_table)
-
-app = FastAPI()
-
-# models.Base.metadata.create_all(bind=engine)
-
-@app.exception_handler(HTTPException)
-async def unicorn_exception_handler(request: Request, exc: HTTPException):
-  return JSONResponse(
-    status_code=exc.status_code,
-    content={"status": exc.status_code, "data": exc.detail},
-  )
-
-app.include_router(authR.authRouter)
-app.include_router(userR.userRouter)
 
 def main():
-  uvicorn.run(
-    "main:app",
-    port=8000,
-    host="0.0.0.0",
-    reload=True
-  )
+  uvicorn.run("app:app", host=settings.API_HOST, port=settings.API_PORT, reload=True, workers=1, forwarded_allow_ips=settings.HOST_IP)
 
-if __name__ == '__main__':
+if __name__ == "__main__":
   main()

fuware/repos/__init__.py

@@ -0,0 +1 @@
+from .repository_users import *

fuware/repos/repository_users.py

@@ -0,0 +1,28 @@
+from fuware.core.security.hasher import get_hasher
+from fuware.db.models.users.users import User
+from fuware.schemas import UserCreate
+from sqlalchemy.orm import Session
+
+
+class RepositoryUsers:
+  def __init__(self):
+    self.user = User()
+
+  def get_all(self, skip: int = 0, limit: int = 100):
+    return self.user.query.offset(skip).limit(limit).all()
+
+  def get_by_username(self, username: str):
+    return self.user.query.filter_by(username=username).first()
+
+  def create(self, db: Session, user: UserCreate):
+    try:
+      hasher = get_hasher()
+      db_user = User(username=user.username, password=hasher.hash(user.password), name=user.name)
+      db.add(db_user)
+      db.commit()
+    except Exception:
+      db.rollback()
+      raise
+
+    db.refresh(db_user)
+    return db_user

fuware/routes/__init__.py

@@ -0,0 +1,7 @@
+from fastapi import APIRouter
+
+from . import (auth)
+
+router = APIRouter(prefix='/api')
+
+router.include_router(auth.router)

fuware/routes/_base/routers.py

@@ -0,0 +1,9 @@
+from enum import Enum
+from fastapi import APIRouter, Depends
+
+from fuware.core.dependencies import get_auth_user
+
+
+class PrivateAPIRouter(APIRouter):
+  def __init__(self, tags: list[str | Enum] | None = None, prefix: str = "", **kwargs):
+    super().__init__(tags=tags, prefix=prefix, dependencies=[Depends(get_auth_user)], **kwargs)

fuware/routes/auth/__init__.py

@@ -0,0 +1,7 @@
+
+from fastapi import APIRouter
+from . import auth
+
+router = APIRouter(prefix='/auth')
+
+router.include_router(auth.public_router)

fuware/routes/auth/auth.py

@@ -0,0 +1,35 @@
+from typing import Any
+from fastapi import APIRouter, Depends, HTTPException, Response
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy.orm import Session
+from fuware.db.db_setup import generate_session
+from fuware.schemas import ReturnValue, PrivateUser, UserRequest
+from fuware.schemas.user.user import UserCreate
+from fuware.services import UserService
+
+
+public_router = APIRouter(tags=["Users: Authentication"])
+
+user_service = UserService()
+
+@public_router.put('/register')
+def register_user(user: UserCreate, db: Session = Depends(generate_session)) -> ReturnValue[Any]:
+  db_user = user_service.get_by_username(username=user.username)
+  if db_user:
+    raise HTTPException(status_code=400, detail="Username already registered!")
+  user_return = user_service.create(db=db, user=user)
+  return ReturnValue(status=200, data=jsonable_encoder(user_return))
+
+# @public_router.post('/login', response_model=ReturnValue[PrivateUser])
+# def user_login(user: UserRequest, response: Response) -> ReturnValue[Any]:
+#   db_user = UserService.get_by_username(user.username)
+#   if not db_user:
+#     raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
+  # if not verify_password(user.password, db_user.password):
+    # raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
+  # if db_user.is_lock is True:
+    # raise HTTPException(status_code=401, detail="Your Account is banned")
+  # cookieEncode = encryptString(user.username + ',' + user.password)
+  # response.set_cookie(key=COOKIE_KEY, value=cookieEncode.decode('utf-8'))
+  # return ReturnValue(status=200, data=jsonable_encoder(db_user))

fuware/routes/authR.py

@@ -1,50 +0,0 @@
-from typing import Any
-from fastapi import APIRouter, HTTPException, Response, Request, Depends
-from fastapi.encoders import jsonable_encoder
-from schemas import ReturnValue, User, UserCreate, UserRequest
-from ultis import root_api_path_build, encryptString, decryptString, verify_password
-from const import COOKIE_KEY
-from sqlalchemy.orm import Session
-from db.controller import get_user_by_username, create_user
-from db import get_db
-
-authRouter=APIRouter(prefix=root_api_path_build('/auth'))
-
-@authRouter.put('/register')
-def register_user(user: UserCreate, db: Session = Depends(get_db)) -> ReturnValue[Any]:
-  db_user = get_user_by_username(db=db, usn=user.username)
-  if db_user:
-    raise HTTPException(status_code=400, detail="Username already registered!")
-  user_return = create_user(db=db, user=user)
-  return ReturnValue(status=200, data=jsonable_encoder(user_return))
-
-@authRouter.post('/login', response_model=ReturnValue[User])
-def user_login(user: UserRequest, response: Response, db: Session = Depends(get_db)) -> ReturnValue[Any]:
-  db_user = get_user_by_username(db, user.username)
-  if not db_user:
-    raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
-  if not verify_password(user.password, db_user.password):
-    raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
-  if db_user.is_lock is True:
-    raise HTTPException(status_code=401, detail="Your Account is banned")
-  cookieEncode = encryptString(user.username + ',' + user.password)
-  response.set_cookie(key=COOKIE_KEY, value=cookieEncode.decode('utf-8'))
-  return ReturnValue(status=200, data=jsonable_encoder(db_user))
-
-@authRouter.get('/logout')
-def user_logout(response: Response) -> ReturnValue[Any]:
-  response.delete_cookie(key=COOKIE_KEY)
-  return ReturnValue(status=200, data='Logged out')
-
-def get_auth_user(request: Request, db: Session = Depends(get_db)):
-  """verify that user has a valid session"""
-  session_id = request.cookies.get(COOKIE_KEY)
-  if not session_id:
-    raise HTTPException(status_code=401, detail="Unauthorized")
-  decrypt_user = decryptString(session_id).split(',')
-  db_user = get_user_by_username(db, decrypt_user[0])
-  if not db_user:
-    raise HTTPException(status_code=403)
-  if not verify_password(decrypt_user[1], db_user.password):
-    raise HTTPException(status_code=401, detail="Your username or password input is wrong!")
-  return True

fuware/routes/userR.py

@@ -1,11 +0,0 @@
-from typing import Any
-from fastapi import APIRouter, Depends
-from schemas import ReturnValue
-from ultis import root_api_path_build
-from routes import authR
-
-userRouter=APIRouter(prefix=root_api_path_build('/user'))
-
-@userRouter.get('/get-data/', dependencies=[Depends(authR.get_auth_user)])
-def get_data(url: str = '') -> ReturnValue[Any]:
-  return ReturnValue(status=200, data=url)

fuware/schemas/fuware_model.py

@@ -0,0 +1,27 @@
+from typing import ClassVar, Protocol, TypeVar
+from humps import camelize
+from enum import Enum
+from pydantic import UUID4, BaseModel, ConfigDict
+
+T = TypeVar("T", bound=BaseModel)
+
+class SearchType(Enum):
+    fuzzy = "fuzzy"
+    tokenized = "tokenized"
+
+class FuwareModel(BaseModel):
+  _searchable_properties: ClassVar[list[str]] = []
+  """
+    Searchable properties for the search API.
+    The first property will be used for sorting (order_by)
+    """
+  model_config = ConfigDict(alias_generator=camelize, populate_by_name=True)
+
+  def cast(self, cls: type[T], **kwargs) -> T:
+    """
+    Cast the current model to another with additional arguments. Useful for
+    transforming DTOs into models that are saved to a database
+    """
+    create_data = {field: getattr(self, field) for field in self.__fields__ if field in cls.__fields__}
+    create_data.update(kwargs or {})
+    return cls(**create_data)

fuware/schemas/user/user.py

@@ -1,8 +1,10 @@
 from datetime import datetime
-from pydantic import BaseModel
+from pydantic import BaseModel, ConfigDict
 from fastapi import Form
 
-class UserBase(BaseModel):
+from fuware.schemas.fuware_model import FuwareModel
+
+class UserBase(FuwareModel):
   username: str = Form(...)
 
 class UserRequest(UserBase):
@@ -12,13 +14,11 @@ class UserCreate(UserRequest):
   password: str = Form(...)
   name: str
 
-class User(UserBase):
+class PrivateUser(UserBase):
   id: str
   name: str
   is_admin: bool
   is_lock: bool
   created_at: datetime
   updated_at: datetime
-
-  class Config:
-    from_attributes = True
+  model_config = ConfigDict(from_attributes=True)

fuware/services/__init__.py

@@ -0,0 +1 @@
+from .user import *

fuware/services/_base_service/__init__.py

@@ -0,0 +1,6 @@
+from fuware.core.config import get_app_settings
+
+
+class BaseService:
+  def __init__(self) -> None:
+    self.setting = get_app_settings()

fuware/services/user/__init__.py

@@ -0,0 +1 @@
+from .user_service import *

fuware/services/user/user_service.py

@@ -0,0 +1,19 @@
+
+from sqlalchemy.orm import Session
+from fuware.repos import RepositoryUsers
+from fuware.services._base_service import BaseService
+from fuware.schemas import UserCreate
+
+
+class UserService(BaseService):
+  def __init__(self):
+    self.repos = RepositoryUsers()
+
+  def get_all(self, skip: int = 0, limit: int = 100):
+    return self.repos.get_all(skip=skip, limit=limit)
+
+  def get_by_username(self, username: str):
+    return self.repos.get_by_username(username)
+
+  def create(self, db: Session, user: UserCreate):
+    return self.repos.create(db=db, user=user)