JLey/FastAPI---Microservices
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

adding .unix

Browse Source
This commit is contained in:
Juan M. Ley
2026-02-07 02:17:02 -06:00
parent afc7aa99c0
commit ea7144c3eb
15 changed files with 270 additions and 270 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
README.txt
View File

@@ -1,31 +1,31 @@
I was too lazy to write this in a markdown file. And lazy enough to write this in English as well.
------------------------------------------------
Welp. Easy RESTful API and stuff.
Endpoints:
- GET | / | Sample root endpoint.
- GET | /docs | Swagger UI for testing the endpoints.
- GET | /users | List all users.
- GET | /users/{id} | Gets a user by its id in database.
- POST | /users | Adds a new user to the database.
Dependencies JIC you want to make your own virtual python environment:
- uvicorn | To run API apps
- fastapi | This project's API Framework
- pydantic | Data validation library
- sqlalchemy | Framework to operate databases with abstractions
Extra implementations:
- Database | SQLite3 Database file. /src/users.db
- Windows & UNIX| For Windows or Linux servers, this API includes both .env directories for windows and UNIX. /.env & /.env-unix
Tested on:
- Windows
- Linux (WSL & Native) - Using Arch Linux.
Not tested on:
- BSD - FreeBSD and/or NetBSD.
- Docker.
- Micropython - ESP32.
I was too lazy to write this in a markdown file. And lazy enough to write this in English as well.
------------------------------------------------
Welp. Easy RESTful API and stuff.
Endpoints:
- GET | / | Sample root endpoint.
- GET | /docs | Swagger UI for testing the endpoints.
- GET | /users | List all users.
- GET | /users/{id} | Gets a user by its id in database.
- POST | /users | Adds a new user to the database.
Dependencies JIC you want to make your own virtual python environment:
- uvicorn | To run API apps
- fastapi | This project's API Framework
- pydantic | Data validation library
- sqlalchemy | Framework to operate databases with abstractions
Extra implementations:
- Database | SQLite3 Database file. /src/users.db
- Windows & UNIX| For Windows or Linux servers, this API includes both .env directories for windows and UNIX. /.env & /.env-unix
Tested on:
- Windows
- Linux (WSL & Native) - Using Arch Linux.
Not tested on:
- BSD - FreeBSD and/or NetBSD.
- Docker.
- Micropython - ESP32.
- Other Python compatible systems.

14
deps.json
View File

@@ -1,8 +1,8 @@
{
"dependencies": {
"1" : "uviconrn",
"2" : "fastapi",
"3" : "pydantic",
"4" : "sqlalchemy"
}
{
"dependencies": {
"1" : "uviconrn",
"2" : "fastapi",
"3" : "pydantic",
"4" : "sqlalchemy"
}
}

4
src/application/exceptions.py
View File

@@ -1,2 +1,2 @@
class UserAlreadyExists(Exception):
pass
class UserAlreadyExists(Exception):
pass

24
src/application/ports/user_repository.py
View File

@@ -1,13 +1,13 @@
from abc import ABC, abstractmethod
from domain.users import User
class UserRepository(ABC):
@abstractmethod
def save(self, user: User):
...
@abstractmethod
def viewAll(self) -> list[User]:
...
@abstractmethod
def viewById(self, user_id:int):
from abc import ABC, abstractmethod
from domain.users import User
class UserRepository(ABC):
@abstractmethod
def save(self, user: User):
...
@abstractmethod
def viewAll(self) -> list[User]:
...
@abstractmethod
def viewById(self, user_id:int):
...

64
src/application/services/user_services.py
View File

@@ -1,33 +1,33 @@
from domain.users import User
from application.ports.user_repository import UserRepository
class CreateUser:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self, name:str, email:str, phone:str):
user = User(user_id=0, email=email, name=name, phone=phone)
self.repo.save(user)
return user
class ViewUsers:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self):
users = self.repo.viewAll()
return users
class ViewUserById:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self, user_id: int) -> User | None:
user = self.repo.viewById(user_id)
from domain.users import User
from application.ports.user_repository import UserRepository
class CreateUser:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self, name:str, email:str, phone:str):
user = User(user_id=0, email=email, name=name, phone=phone)
self.repo.save(user)
return user
class ViewUsers:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self):
users = self.repo.viewAll()
return users
class ViewUserById:
def __init__(self, repo: UserRepository):
if not isinstance(repo, UserRepository):
raise TypeError("repo must implement UserRepository")
self.repo = repo
def execute(self, user_id: int) -> User | None:
user = self.repo.viewById(user_id)
return user

14
src/domain/users.py
View File

@@ -1,8 +1,8 @@
from dataclasses import dataclass
@dataclass
class User:
user_id: int
name: str
email: str
from dataclasses import dataclass
@dataclass
class User:
user_id: int
name: str
email: str
phone: str

10
src/infrastructure/adapters/persistence/__init__.py
View File

@@ -1,5 +1,5 @@
from infrastructure.adapters.persistence.db import engine
from infrastructure.adapters.persistence.models import Base
Base.metadata.create_all(bind=engine)
from infrastructure.adapters.persistence.db import engine
from infrastructure.adapters.persistence.models import Base
Base.metadata.create_all(bind=engine)

32
src/infrastructure/adapters/persistence/db.py
View File

@@ -1,17 +1,17 @@
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False
)
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, declarative_base
DATABASE_URL = "sqlite:///./users.db"
engine = create_engine(
DATABASE_URL,
connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(
bind=engine,
autocommit=False,
autoflush=False
)
Base = declarative_base()

18
src/infrastructure/adapters/persistence/models.py
View File

@@ -1,10 +1,10 @@
from sqlalchemy import Column, Integer, String
from infrastructure.adapters.persistence.db import Base
class UserModel(Base):
__tablename__ = "users"
user_id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=False, index=True)
email = Column(String, unique=True, index=True)
from sqlalchemy import Column, Integer, String
from infrastructure.adapters.persistence.db import Base
class UserModel(Base):
__tablename__ = "users"
user_id = Column(Integer, primary_key=True, index=True)
name = Column(String, unique=False, index=True)
email = Column(String, unique=True, index=True)
phone = Column(String, unique=True, index=True)

120
src/infrastructure/adapters/persistence/user_repository_sql.py
View File

@@ -1,61 +1,61 @@
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from domain.users import User
from application.ports.user_repository import UserRepository
from infrastructure.adapters.persistence.db import SessionLocal
from infrastructure.adapters.persistence.models import UserModel
from application.exceptions import UserAlreadyExists
class SqlUserRepository(UserRepository):
def __init__(self):
self.db: Session = SessionLocal()
def save(self, user: User):
model = UserModel(
name= user.name,
email=user.email,
phone=user.phone
)
self.db.add(model)
try:
self.db.commit()
except IntegrityError:
self.db.rollback()
raise UserAlreadyExists("A user with the same phone number or name does already exist.")
self.db.refresh(model)
return User(
user_id = model.user_id,
name = model.name,
email = model.email,
phone = model.phone
)
def viewAll(self) -> list[User]:
models = self.db.query(UserModel).all()
return [
User(
user_id=model.user_id,
name = model.name,
email = model.email,
phone = model.phone
) for model in models
]
def viewById(self, user_id: int) -> User | None:
model = self.db.query(UserModel).filter(UserModel.user_id == user_id).first()
if not model:
return None
return User(
user_id=model.user_id,
name = model.name,
email = model.email,
phone = model.phone
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from domain.users import User
from application.ports.user_repository import UserRepository
from infrastructure.adapters.persistence.db import SessionLocal
from infrastructure.adapters.persistence.models import UserModel
from application.exceptions import UserAlreadyExists
class SqlUserRepository(UserRepository):
def __init__(self):
self.db: Session = SessionLocal()
def save(self, user: User):
model = UserModel(
name= user.name,
email=user.email,
phone=user.phone
)
self.db.add(model)
try:
self.db.commit()
except IntegrityError:
self.db.rollback()
raise UserAlreadyExists("A user with the same phone number or name does already exist.")
self.db.refresh(model)
return User(
user_id = model.user_id,
name = model.name,
email = model.email,
phone = model.phone
)
def viewAll(self) -> list[User]:
models = self.db.query(UserModel).all()
return [
User(
user_id=model.user_id,
name = model.name,
email = model.email,
phone = model.phone
) for model in models
]
def viewById(self, user_id: int) -> User | None:
model = self.db.query(UserModel).filter(UserModel.user_id == user_id).first()
if not model:
return None
return User(
user_id=model.user_id,
name = model.name,
email = model.email,
phone = model.phone
)

12
src/infrastructure/api/users/app.py
View File

@@ -1,7 +1,7 @@
from fastapi import FastAPI
from infrastructure.api.users.router import router
def create_app() -> FastAPI:
app = FastAPI(title="Users service")
app.include_router(router)
from fastapi import FastAPI
from infrastructure.api.users.router import router
def create_app() -> FastAPI:
app = FastAPI(title="Users service")
app.include_router(router)
return app

18
src/infrastructure/api/users/root.py
View File

@@ -1,10 +1,10 @@
from fastapi import APIRouter
router = APIRouter()
@router.get("/")
def root():
return {
"Status": "Running",
"Docs": "/docs"
from fastapi import APIRouter
router = APIRouter()
@router.get("/")
def root():
return {
"Status": "Running",
"Docs": "/docs"
}

32
src/infrastructure/api/users/router.py
View File

@@ -1,17 +1,17 @@
from fastapi import APIRouter
from infrastructure.api.users.users import router as users_router
from infrastructure.api.users.root import router as root_router
router = APIRouter()
router.include_router(
users_router,
prefix="/users",
tags=["users"]
)
router.include_router(
root_router,
prefix='',
tags=["root"]
from fastapi import APIRouter
from infrastructure.api.users.users import router as users_router
from infrastructure.api.users.root import router as root_router
router = APIRouter()
router.include_router(
users_router,
prefix="/users",
tags=["users"]
)
router.include_router(
root_router,
prefix='',
tags=["root"]
)

86
src/infrastructure/api/users/users.py
View File

@@ -1,44 +1,44 @@
from fastapi import APIRouter, HTTPException
from application.services.user_services import CreateUser, ViewUsers, ViewUserById
from infrastructure.adapters.persistence.user_repository_sql import SqlUserRepository
from application.exceptions import UserAlreadyExists
router = APIRouter()
@router.post("/")
def create_user(name:str, email:str, phone:str):
service = CreateUser(SqlUserRepository())
try:
return service.execute(name,email,phone)
except UserAlreadyExists as e:
raise HTTPException (
status_code=409,
detail=str(e)
)
@router.get("/")
def view_all_users():
service = ViewUsers(SqlUserRepository())
try:
return service.execute()
except Exception as e:
raise HTTPException(
status_code=500,
detail=e)
@router.get("/{id}")
def view_user_by_id(user_id: int):
service = ViewUserById(SqlUserRepository())
result = service.execute(user_id)
if result is None:
raise HTTPException(
status_code=404,
detail="User not found")
from fastapi import APIRouter, HTTPException
from application.services.user_services import CreateUser, ViewUsers, ViewUserById
from infrastructure.adapters.persistence.user_repository_sql import SqlUserRepository
from application.exceptions import UserAlreadyExists
router = APIRouter()
@router.post("/")
def create_user(name:str, email:str, phone:str):
service = CreateUser(SqlUserRepository())
try:
return service.execute(name,email,phone)
except UserAlreadyExists as e:
raise HTTPException (
status_code=409,
detail=str(e)
)
@router.get("/")
def view_all_users():
service = ViewUsers(SqlUserRepository())
try:
return service.execute()
except Exception as e:
raise HTTPException(
status_code=500,
detail=e)
@router.get("/{id}")
def view_user_by_id(user_id: int):
service = ViewUserById(SqlUserRepository())
result = service.execute(user_id)
if result is None:
raise HTTPException(
status_code=404,
detail="User not found")
return result

32
src/main.py
View File

@@ -1,16 +1,16 @@
from infrastructure.api.users.app import create_app
app = create_app()
def run():
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info",
)
if __name__ == "__main__":
run()
from infrastructure.api.users.app import create_app
app = create_app()
def run():
import uvicorn
uvicorn.run(
"main:app",
host="0.0.0.0",
port=8000,
reload=True,
log_level="info",
)
if __name__ == "__main__":
run()