Commit 1dd44468 authored by porramet's avatar porramet

first commit

parents
Pipeline #336 failed with stages
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# 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
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/#use-with-ide
.pdm.toml
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
\ No newline at end of file
RUN
```
uvicorn src.main:app --reload
```
\ No newline at end of file
File added
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db, user=user)
@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
@app.post("/users/{user_id}/items/", response_model=schemas.Item)
def create_item_for_user(
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
):
return crud.create_user_item(db=db, item=item, user_id=user_id)
@app.get("/items/", response_model=list[schemas.Item])
def read_items(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
items = crud.get_items(db, skip=skip, limit=limit)
return items
from sqlalchemy.orm import Session
from . import models, schemas
def create_fake_items():
items= []
for i in range(5):
item = {
models.Item.title.name : f"Item {i+1}",
models.Item.description.name : f"Item Description {i+1}"
}
items.append(item)
items = [models.Item(**item) for item in items]
return items
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
fake_hashed_password = user.password + "notreallyhashed"
db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
db_user.items.extend(create_fake_items())
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
def get_items(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Item).offset(skip).limit(limit).all()
def create_user_item(db: Session, item: schemas.ItemCreate, user_id: int):
db_item = models.Item(**item.dict(), owner_id=user_id)
db.add(db_item)
db.commit()
db.refresh(db_item)
return db_item
def change_image(db:Session):
image = db.query(models.Image).first()
if image.url == 1 :
image.url = 2
else :
image.url = 1
db.commit()
def get_image(db:Session):
image = db.query(models.Image).first()
if image is None :
image = models.Image(url=1)
db.add(image)
db.commit()
db.refresh(image)
return image
\ No newline at end of file
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = "sqlite:///./sql_app.db"
# SQLALCHEMY_DATABASE_URL = "postgresql://user:password@postgresserver/db"
engine = create_engine(
SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
import os
import pandas as pd
import io
from pydantic import Field
from fastapi.responses import PlainTextResponse,RedirectResponse,FileResponse,StreamingResponse
from fastapi_pagination import Page, add_pagination, paginate
from sqlalchemy import select
from . import crud, models, schemas
from .database import SessionLocal, engine
models.Base.metadata.create_all(bind=engine)
app = FastAPI()
Page = Page.with_custom_options(
size=Field(5,ge=1,le=1000)
)
# Dependency
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db, user=user)
@app.post("/create-fake-users")
def create_fake_users(db:Session=Depends(get_db)):
users = []
for i in range(10) :
dd = {
"email" : f"fake{i+1}@example.com",
"password" : f"fake{i+1}@pass"
}
user = schemas.UserCreate(**dd)
uu = crud.create_user(db=db,user=user)
users.append(uu)
return users
@app.get("/users/", response_model=list[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
@app.get('/users-page')
def get_users(page: int = 1, page_size: int = 10, db: Session = Depends(get_db)):
# Calculate the offset for pagination
offset = (page - 1) * page_size
# Fetch users from the database using SQLAlchemy query
users = db.query(models.User).offset(offset).limit(page_size).all()
# Get the total count of users
total_users = db.query(models.User).count()
# Calculate the next and previous page numbers
next_page = page + 1 if (offset + page_size) < total_users else None
previous_page = page - 1 if page > 1 else None
# Construct the response
response = {
"users": users,
"total": total_users,
"page": page,
"next_page": next_page,
"previous_page": previous_page,
}
# Return the response
return response
@app.get('/',include_in_schema=False)
def docs_redirect():
docs_url = f"/docs"
return RedirectResponse(url=docs_url)
@app.get("/text",response_class=PlainTextResponse)
def return_text():
return "Hello World"
@app.get("/image",response_class=FileResponse)
def return_image(db:Session= Depends(get_db)) :
image = crud.get_image(db)
return os.path.join("static","images",f"{image.url}.jpg")
@app.get('/change/image')
def change_image(db:Session= Depends(get_db)) :
crud.change_image(db)
return "OK"
@app.get('/csv')
def return_csv (db:Session= Depends(get_db)) :
df = pd.read_sql(
sql= db.query(models.User).statement,
con =engine
)
stream = io.StringIO()
df.to_csv(stream,index=False)
response = StreamingResponse(iter([stream.getvalue()]),media_type="text/csv")
response.headers["Content-Disposition"] = "attachment; filename=export.csv"
return response
@app.get("/excel")
def return_excel (db:Session=Depends(get_db)):
df = pd.read_sql(
sql= db.query(models.User).statement,
con =engine
)
stream = io.BytesIO()
df.to_excel(stream,index=False)
response = StreamingResponse(iter([stream.getvalue()]),media_type="application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")
response.headers["Content-Disposition"] = "attachment; filename=export.xlsx"
return response
add_pagination(app)
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from .database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=False, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
items = relationship("Item", back_populates="owner")
class Item(Base):
__tablename__ = "items"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
description = Column(String, index=True)
owner_id = Column(Integer, ForeignKey("users.id"))
owner = relationship("User", back_populates="items")
class Image(Base):
__tablename__ = "image"
id = Column(Integer,primary_key=True)
url = Column(Integer)
\ No newline at end of file
from pydantic import BaseModel
class ItemBase(BaseModel):
title: str
description: str | None = None
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
owner_id: int
class Config:
from_attributes = True
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
items: list[Item] = []
class Config:
from_attributes = True
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment