Add and delete files, update dependencies, and refactor code

bemore/__main__.py

@@ -1,7 +1,7 @@
 import uvicorn
 
 from bemore.gunicorn_runner import GunicornApplication
-from bemore.settings import settings
+from bemore.core.config import settings
 
 
 def main() -> None:

bemore/settings.py → bemore/core/config.py

@@ -1,11 +1,6 @@
 import enum
-from pathlib import Path
-from tempfile import gettempdir
-
+import secrets
 from pydantic_settings import BaseSettings, SettingsConfigDict
-from yarl import URL
-
-TEMP_DIR = Path(gettempdir())
 
 
 class LogLevel(str, enum.Enum):  # noqa: WPS600
@@ -27,46 +22,29 @@ class Settings(BaseSettings):
     with environment variables.
     """
 
+    API_STR: str = "/api"
+    SECRET_KEY: str = secrets.token_urlsafe(32)
     host: str = "127.0.0.1"
     port: int = 8000
     # quantity of workers for uvicorn
     workers_count: int = 1
     # Enable uvicorn reloading
-    reload: bool = True
+    reload: bool = False
 
     # Current environment
     environment: str = "dev"
 
     log_level: LogLevel = LogLevel.INFO
-    # Variables for the database
-    db_host: str = "localhost"
-    db_port: int = 3306
-    db_user: str = "bemore"
-    db_pass: str = "bemore"
-    db_base: str = "bemore"
-    db_echo: bool = False
-
-    @property
-    def db_url(self) -> URL:
-        """
-        Assemble database URL from settings.
-
-        :return: database URL.
-        """
-        return URL.build(
-            scheme="mysql",
-            host=self.db_host,
-            port=self.db_port,
-            user=self.db_user,
-            password=self.db_pass,
-            path=f"/{self.db_base}",
-        )
 
     model_config = SettingsConfigDict(
         env_file=".env",
         env_prefix="BEMORE_",
         env_file_encoding="utf-8",
     )
 
+    SQLALCHEMY_DATABASE_URI: str = "sqlite:///./bemore.db"
+    FIRST_SUPERUSER: str = "admin@localhost"
+    FIRST_SUPERUSER_PASSWORD: str = "admin"
+
 
 settings = Settings()

bemore/core/security.py

@@ -0,0 +1,34 @@
+from datetime import datetime, timedelta
+from typing import Any, Union
+
+from jose import jwt
+from passlib.context import CryptContext
+
+from bemore.core import config
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+ALGORITHM = "HS256"
+
+
+def create_access_token(
+    subject: Union[str, Any], expires_delta: timedelta = None
+) -> str:
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(
+            minutes=config.ACCESS_TOKEN_EXPIRE_MINUTES
+        )
+    to_encode = {"exp": expire, "sub": str(subject)}
+    encoded_jwt = jwt.encode(to_encode, config.SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def verify_password(plain_password: str, hashed_password: str) -> bool:
+    return pwd_context.verify(plain_password, hashed_password)
+
+
+def get_password_hash(password: str) -> str:
+    return pwd_context.hash(password)

bemore/crud/base.py

@@ -0,0 +1,59 @@
+from typing import Any, Dict, Generic, Optional, Type, TypeVar, Union
+
+from fastapi.encoders import jsonable_encoder
+from pydantic import BaseModel
+from sqlalchemy.orm import Session
+
+ModelType = TypeVar("ModelType", bound=Any)
+CreateSchemaType = TypeVar("CreateSchemaType", bound=BaseModel)
+UpdateSchemaType = TypeVar("UpdateSchemaType", bound=BaseModel)
+
+
+class CRUDBase(Generic[ModelType, CreateSchemaType, UpdateSchemaType]):
+    def __init__(self, model: Type[ModelType]):
+        """
+        CRUD object with default methods to Create, Read, Update, Delete (CRUD).
+
+        **Parameters**
+
+        * `model`: A SQLAlchemy model class
+        * `schema`: A Pydantic model (schema) class
+        """
+        self.model = model
+
+    def get(self, db: Session, id: Any) -> Optional[ModelType]:
+        return db.query(self.model).filter(self.model.id == id).first()
+
+    def create(self, db: Session, *, obj_in: CreateSchemaType) -> ModelType:
+        obj_in_data = jsonable_encoder(obj_in)
+        db_obj = self.model(**obj_in_data)  # type: ignore
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    def update(
+        self,
+        db: Session,
+        *,
+        db_obj: ModelType,
+        obj_in: Union[UpdateSchemaType, Dict[str, Any]]
+    ) -> ModelType:
+        obj_data = jsonable_encoder(db_obj)
+        if isinstance(obj_in, dict):
+            update_data = obj_in
+        else:
+            update_data = obj_in.dict(exclude_unset=True)
+        for field in obj_data:
+            if field in update_data:
+                setattr(db_obj, field, update_data[field])
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    def remove(self, db: Session, *, id: int) -> ModelType:
+        obj = db.query(self.model).get(id)
+        db.delete(obj)
+        db.commit()
+        return obj

bemore/crud/crud_item.py

@@ -0,0 +1,34 @@
+from typing import List
+
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy.orm import Session
+
+from bemore.crud.base import CRUDBase
+from bemore.models import Item
+from bemore.schemas.item import ItemCreate, ItemUpdate
+
+
+class CRUDItem(CRUDBase[Item, ItemCreate, ItemUpdate]):
+    def create_with_owner(
+        self, db: Session, *, obj_in: ItemCreate, owner_id: int
+    ) -> Item:
+        obj_in_data = jsonable_encoder(obj_in)
+        db_obj = self.model(**obj_in_data, owner_id=owner_id)
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    def get_multi_by_owner(
+        self, db: Session, *, owner_id: int, skip: int = 0, limit: int = 100
+    ) -> List[Item]:
+        return (
+            db.query(self.model)
+            .filter(Item.owner_id == owner_id)
+            .offset(skip)
+            .limit(limit)
+            .all()
+        )
+
+
+item = CRUDItem(Item)

bemore/crud/crud_user.py

@@ -0,0 +1,55 @@
+from typing import Any, Dict, Optional, Union
+
+from sqlalchemy.orm import Session
+
+from bemore.core.security import get_password_hash, verify_password
+from bemore.crud.base import CRUDBase
+from bemore.models import User
+from bemore.schemas.user import UserCreate, UserUpdate
+
+
+class CRUDUser(CRUDBase[User, UserCreate, UserUpdate]):
+    def get_by_email(self, db: Session, *, email: str) -> Optional[User]:
+        return db.query(User).filter(User.email == email).first()
+
+    def create(self, db: Session, *, obj_in: UserCreate) -> User:
+        db_obj = User(
+            email=obj_in.email,
+            hashed_password=get_password_hash(obj_in.password),
+            full_name=obj_in.full_name,
+            is_superuser=obj_in.is_superuser,
+        )
+        db.add(db_obj)
+        db.commit()
+        db.refresh(db_obj)
+        return db_obj
+
+    def update(
+        self, db: Session, *, db_obj: User, obj_in: Union[UserUpdate, Dict[str, Any]]
+    ) -> User:
+        if isinstance(obj_in, dict):
+            update_data = obj_in
+        else:
+            update_data = obj_in.model_dump(exclude_unset=True)
+        if update_data["password"]:
+            hashed_password = get_password_hash(update_data["password"])
+            del update_data["password"]
+            update_data["hashed_password"] = hashed_password
+        return super().update(db, db_obj=db_obj, obj_in=update_data)
+
+    def authenticate(self, db: Session, *, email: str, password: str) -> Optional[User]:
+        user = self.get_by_email(db, email=email)
+        if not user:
+            return None
+        if not verify_password(password, user.hashed_password):
+            return None
+        return user
+
+    def is_active(self, user: User) -> bool:
+        return user.is_active
+
+    def is_superuser(self, user: User) -> bool:
+        return user.is_superuser
+
+
+user = CRUDUser(User)

bemore/db/engine.py

@@ -0,0 +1,5 @@
+from sqlmodel import create_engine
+
+from bemore.core.config import settings
+
+engine = create_engine(settings.SQLALCHEMY_DATABASE_URI)

bemore/db/init_db.py

@@ -0,0 +1,26 @@
+from sqlmodel import Session, select
+
+from bemore.crud.crud_user import user as crud
+from bemore.core.config import settings
+from bemore.models import User, UserCreate  # noqa: F401
+
+# make sure all SQLModel models are imported (app.models) before initializing DB
+# otherwise, SQLModel might fail to initialize relationships properly
+# for more details: https://github.com/tiangolo/full-stack-fastapi-postgresql/issues/28
+
+
+def init_db(session: Session) -> None:
+    # Tables should be created with Alembic migrations
+    # But if you don't want to use migrations, create
+    # the tables un-commenting the next line
+    # Base.metadata.create_all(bind=engine)
+    user = session.exec(
+        select(User).where(User.email == settings.FIRST_SUPERUSER)
+    ).first()
+    if not user:
+        user_in = UserCreate(
+            email=settings.FIRST_SUPERUSER,
+            password=settings.FIRST_SUPERUSER_PASSWORD,
+            is_superuser=True,
+        )
+        user = crud.create(db=session, obj_in=user_in)

bemore/initial_data.py

@@ -0,0 +1,17 @@
+from sqlmodel import Session
+
+from bemore.db.engine import engine
+from bemore.db.init_db import init_db
+
+
+def init() -> None:
+    with Session(engine) as session:
+        init_db(session)
+
+
+def main() -> None:
+    init()
+
+
+if __name__ == "__main__":
+    main()

bemore/logging.py

@@ -4,7 +4,7 @@
 
 from loguru import logger
 
-from bemore.settings import settings
+from bemore.core.config import settings
 
 
 class InterceptHandler(logging.Handler):