Skip to content

F
financial-system
Commit 1095519a authored by xianyang's avatar xianyang
Browse files
Options

优化层级目录

parent 750caa25
Hide whitespace changes
Inline Side-by-side
Showing with 50 additions and 297 deletions
app/__init__.py
View file @ 1095519a
import pymysql import pymysql
pymysql.install_as_MySQLdb() pymysql.install_as_MySQLdb()
\ No newline at end of file
app/api/api_v1.py 0 → 100644
View file @ 1095519a
from fastapi import APIRouter
from app.api.users import views
api_router = APIRouter()
# api_router.include_router(login.router, tags=["login"])
# api_router.include_router(items.router, prefix="/items", tags=["items"])
api_router.include_router(views.router, prefix="/users", tags=["users"])
app/users/router.py → app/api/users/views.py
View file @ 1095519a
from typing import List from typing import List
from fastapi import Depends, APIRouter, HTTPException, FastAPI
from fastapi import Depends, APIRouter, HTTPException
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from app.users import crud, models, schemas, router from app.users import crud, models, schemas
from core.storage.db import SessionLocal, engine from core.storage.db import SessionLocal
from core.dependencies.auth_dependen import check_token
# models.Base.metadata.create_all(bind=engine) # models.Base.metadata.create_all(bind=engine)
router = APIRouter()
# Dependency # Dependency
def get_db(): def get_db():
...@@ -46,3 +46,15 @@ def create_item_for_user( ...@@ -46,3 +46,15 @@ def create_item_for_user(
user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db) user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
): ):
return crud.create_user_item(db=db, item=item, user_id=user_id) return crud.create_user_item(db=db, item=item, user_id=user_id)
# @router.get("/items", response_model=schemas.Item)
# def createm_for_user(
# user_id: int, item: schemas.ItemCreate, db: Session = Depends(get_db)
# ):
# return {'message': 'Hello demo Applications!'}
@router.get("/items", tags=['GET', 'POST'])
def ceshi():
return {'message': 'Hello demo Adsadsadsadsadaspplications!'}
app/demo/crud.py deleted 100644 → 0
View file @ 750caa25
from sqlalchemy.orm import Session
from . import models, schemas
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.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
app/demo/models.py deleted 100644 → 0
View file @ 750caa25
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from ...core.storage.db import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, 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")
\ No newline at end of file
app/demo/router.py deleted 100644 → 0
View file @ 750caa25
from typing import List
from fastapi import Depends, FastAPI, HTTPException
from sqlalchemy.orm import Session
from . import crud, models, schemas
from ...core.storage.db 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.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
app/demo/schemas.py deleted 100644 → 0
View file @ 750caa25
from typing import List, Union
from pydantic import BaseModel
class ItemBase(BaseModel):
title: str
description: Union[str, None] = None
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
owner_id: int
class Config:
orm_mode = True
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
items: List[Item] = []
class Config:
orm_mode = True
app/users/__init__.py deleted 100644 → 0
View file @ 750caa25
from fastapi import Depends, APIRouter
from app.users import crud, models, schemas
from core.storage.db import engine
from core.dependencies.auth_dependen import check_token
models.Base.metadata.create_all(bind=engine)
router = APIRouter(prefix='/users', dependencies=[Depends(check_token)])
app/users/crud.py deleted 100644 → 0
View file @ 750caa25
from sqlalchemy.orm import Session
from app.users import models, schemas
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.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
app/users/models.py deleted 100644 → 0
View file @ 750caa25
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String
from sqlalchemy.orm import relationship
from core.storage.db import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String(50), unique=True, index=True)
hashed_password = Column(String(100))
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")
\ No newline at end of file
app/users/schemas.py deleted 100644 → 0
View file @ 750caa25
from typing import List, Union
from pydantic import BaseModel
class ItemBase(BaseModel):
title: str
description: Union[str, None] = None
class ItemCreate(ItemBase):
pass
class Item(ItemBase):
id: int
owner_id: int
class Config:
orm_mode = True
class UserBase(BaseModel):
email: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
items: List[Item] = []
class Config:
orm_mode = True
core/config/env.py
View file @ 1095519a
...@@ -3,23 +3,34 @@ ...@@ -3,23 +3,34 @@
切换环境，修改文档后面的命令 env = ... 切换环境，修改文档后面的命令 env = ...
""" """
import os
from pydantic import BaseSettings from pydantic import BaseSettings
from starlette.middleware.cors import CORSMiddleware
from main import app
from app.api.api_v1 import api_router
class Env(BaseSettings): class Env(BaseSettings):
DEBUG: bool = False DEBUG: bool = True
TESTING: bool = False TESTING: bool = False
DATABASE_URI: str = 'sqlite://:memory:' DATABASE_URI: str = 'sqlite://:memory:'
DATABASE_USER: str = '' DATABASE_USER: str = ''
DATABASE_PWD: str = '' DATABASE_PWD: str = ''
API_V1_STR = "/api"
PROJECT_NAME = os.getenv("PROJECT_NAME")
# 设置允许访问的域名
origins = ['*'] # 可以设置为'*'，即为所有。
# 设置跨域传参
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 设置允许的origins来源
allow_credentials=True,
allow_methods=['*'], # 设置允许跨域的http方法，比如 get、post、put等。
allow_headers=['*']) # 允许跨域的headers，可以用来鉴别来源等作用。
app.include_router(api_router, prefix=API_V1_STR)
class DevelopmentEnv(Env):
'''
开发环境配置
'''
DEBUG: bool = True
class TestingEnv(Env): class TestingEnv(Env):
''' '''
...@@ -27,28 +38,14 @@ class TestingEnv(Env): ...@@ -27,28 +38,14 @@ class TestingEnv(Env):
''' '''
TESTING: bool = True TESTING: bool = True
class PrehubEnv(Env):
'''
预发布环境配置
'''
DATABASE_URI: str = 'mysql://user@localhost/foo'
class ProdEnv(Env): class ProdEnv(Env):
''' '''生产环境配置'''
生产环境配置
'''
DATABASE_URI: str = 'mysql://user@localhost/foo' DATABASE_URI: str = 'mysql://user@localhost/foo'
DATABASE_USER: str = '' DATABASE_USER: str = ''
DATABASE_PWD: str = '' DATABASE_PWD: str = ''
env = DevelopmentEnv() # 开发环境 env = Env() # 开发环境
# env = TestingEnv() # 测试环境 # env = TestingEnv() # 测试环境
# env = PrehubEnv() # 预发布环境
# env = ProdEnv() # 生产环境 # env = ProdEnv() # 生产环境
main.py
View file @ 1095519a
import uvicorn import uvicorn
from fastapi import FastAPI
from fastapi import Depends, FastAPI,Request,Response, HTTPException,status app = FastAPI()
from starlette.middleware.cors import CORSMiddleware
# from core.dependencies.base_dependen import get_query_token, get_token_header if __name__ == '__main__':
# from internal import admin uvicorn.run(app=app, host="127.0.0.1", port=8001)
from app.users.router import router as users_router
from core.dependencies.auth_dependen import router as token_router
app = FastAPI(
title='demo',
description='',
version='1.0.0')
# 包含子模块的路由
app.include_router(users_router, tags=['users'])
# app.include_router(admin.router, prefix='/admin', tags=['admin'] )
app.include_router(token_router, tags=['token'])
# 可选,独立定义根路由
@app.get('/',tags=['root'])
async def root():
return {'message': 'Hello demo Applications!'}
# 设置允许访问的域名
origins = ['*'] #可以设置为'*'，即为所有。
# 设置跨域传参
app.add_middleware(
CORSMiddleware,
allow_origins=origins, # 设置允许的origins来源
allow_credentials=True,
allow_methods=['*'], # 设置允许跨域的http方法，比如 get、post、put等。
allow_headers=['*']) # 允许跨域的headers，可以用来鉴别来源等作用。
# if __name__ == '__main__':
# uvicorn.run(app,debug=True)
\ No newline at end of file
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