отрефакторил и вроде работает
Gitea Actions Demo / Explore-Gitea-Actions (push) Successful in 5s Details

This commit is contained in:
sergey 2024-08-11 20:18:32 +03:00
parent aaa88b7442
commit d6b36ad456
20 changed files with 166 additions and 83 deletions

View File

@ -1,3 +1,7 @@
BLOCKING_IP__DB__URL=postgresql+asyncpg://username:password@localhost:5432/dbname
BLOCKING_IP__DB__drivername=postgresql+asyncpg
BLOCKING_IP__DB__username=username
BLOCKING_IP__DB__password=password
BLOCKING_IP__DB__host=localhost
BLOCKING_IP__DB__port=5432
BLOCKING_IP__DB__database=dbname
BLOCKING_IP__DB__ECHO=0
ALEMBIC_CONFIG=alembic/alembic.ini

View File

@ -20,7 +20,7 @@ if config.config_file_name is not None:
# for 'autogenerate' support
# from myapp import mymodel
# target_metadata = mymodel.Base.metadata
from core.models import Base
from core.models.base import Base
target_metadata = Base.metadata

View File

@ -1,11 +1,10 @@
from api.swagger import router as swagger_router
from fastapi import APIRouter
from core.config import settings
from api.ip import router as ip_router
router = APIRouter()
router.include_router(
ip_router,
prefix=settings.api.ip,
swagger_router,
prefix=settings.prefix.swagger,
)

View File

@ -1,17 +0,0 @@
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from core.models import db_helper
from core.schemas.ip import IpRead
router = APIRouter(
tags=["Ip"],
)
@router.get("", response_model=list[IpRead])
async def get_ip(session: AsyncSession = Depends(db_helper.session_dependency)):
pass

19
api/swagger.py Normal file
View File

@ -0,0 +1,19 @@
from fastapi import APIRouter
from fastapi.openapi.docs import get_swagger_ui_html
from core.config import settings
router = APIRouter()
@router.get("", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=settings.swagger.openapi_url,
title=settings.swagger.title,
oauth2_redirect_url=settings.swagger.oauth2_redirect_url,
swagger_js_url=settings.swagger.swagger_js_url,
swagger_css_url=settings.swagger.swagger_css_url,
swagger_favicon_url=settings.swagger.swagger_favicon_url,
)

View File

@ -5,6 +5,8 @@ from pydantic import BaseModel, PostgresDsn
from pydantic_settings import BaseSettings, SettingsConfigDict
BASE_DIR = Path(__file__).parent.parent
TEMPLATES_DIR = BASE_DIR / "templates"
STATIC_DIR = TEMPLATES_DIR / "static"
class RunConfig(BaseModel):
@ -14,19 +16,39 @@ class RunConfig(BaseModel):
class DatabaseConfig(BaseModel):
url: str
drivername: str = "postgresql+asyncpg"
username: str = "username"
password: str = "password"
host: str = "localhost"
port: int = 5432
database: str = "dbname"
url: PostgresDsn = f"{drivername}://{username}:{password}@{host}:{port}/{database}"
echo: bool = False
echo_pool: bool = False
pool_size: int = 50
max_overflow: int = 10
naming_convention: dict[str, str] = {
"ix": "ix_%(column_0_label)s",
"uq": "uq_%(table_name)s_%(column_0_name)s",
"ck": "ck_%(table_name)s_%(constraint_name)s",
"fk": "fk_%(table_name)s_%(column_0_name)s_%(referred_table_name)s",
"pk": "pk_%(table_name)s",
}
class ApiPrefix(BaseModel):
class PrefixConfig(BaseModel):
api: str = "/api"
ip: str = "/api/ip"
swagger: str = "/docs"
class WebPrefix(BaseModel):
pass
class SwaggerConfig(BaseModel):
openapi_url: str = "/openapi.json"
title: str = "Blocked IP API"
oauth2_redirect_url: str = "/docs/oauth2-redirect"
swagger_js_url: str = "/static/swagger/swagger-ui-bundle.js"
swagger_css_url: str = "/static/swagger/swagger-ui.css"
swagger_favicon_url: str = "/static/swagger/favicon.png"
class LogConfig(BaseModel):
@ -45,13 +67,21 @@ class Settings(BaseSettings):
env_prefix="BLOCKING_IP__",
)
run: RunConfig = RunConfig()
api: ApiPrefix = ApiPrefix()
web: WebPrefix = WebPrefix()
prefix: PrefixConfig = PrefixConfig()
log: LogConfig = LogConfig()
swagger: SwaggerConfig = SwaggerConfig()
db: DatabaseConfig
settings = Settings()
settings.db.url = (
f"{settings.db.drivername}://"
f"{settings.db.username}:"
f"{settings.db.password}@"
f"{settings.db.host}:"
f"{settings.db.port}/"
f"{settings.db.database}"
)
def config_logging(level: str) -> None:

View File

@ -1,9 +1,10 @@
__all__ = (
from core.models.base import Base
from core.models.db_helper import db_helper
from core.models.ip import Ip
__all__ = [
"db_helper",
"Base",
"Ip",
"db_helper",
)
from .base import Base
from .ip import Ip
from .db_helper import DatabaseHelper, db_helper
]

View File

@ -1,15 +1,18 @@
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, declared_attr
from utils import camel_case_to_snake_case
from core.config import settings
class Base(DeclarativeBase):
"""
Базовый класс для описания моделей.
"""
__abstract__ = True # Что бы эта таблица не создавалась в базе данных
__abstract__ = True
metadata = MetaData(
naming_convention=settings.db.naming_convention,
)
@declared_attr.directive
def __tablename__(cls) -> str:
return f"{cls.__name__.lower()}s"
return f"{camel_case_to_snake_case(cls.__name__)}s"
id: Mapped[int] = mapped_column(primary_key=True)

View File

@ -1,43 +1,57 @@
from asyncio import current_task
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import (
AsyncSession,
create_async_engine,
async_sessionmaker,
async_scoped_session,
AsyncEngine,
)
from core.config import settings
class DatabaseHelper:
def __init__(self, url: str, echo: bool = False):
self.engine = create_async_engine(
def __init__(
self,
url: str,
echo: bool,
echo_pool: bool,
pool_size: int,
max_overflow: int,
) -> None:
self.engine: AsyncEngine = create_async_engine(
url=url,
echo=echo,
echo_pool=echo_pool,
pool_size=pool_size,
max_overflow=max_overflow,
)
self.session_factory = async_sessionmaker(
self.session_factory: async_sessionmaker[AsyncSession] = async_sessionmaker(
bind=self.engine,
autoflush=False,
autocommit=False,
expire_on_commit=False,
)
def get_scoped_session(self):
session = async_scoped_session(
session_factory=self.session_factory,
scopefunc=current_task,
)
return session
# def get_scoped_session(self):
# session = async_scoped_session(
# session_factory=self.session_factory,
# scopefunc=current_task,
# )
# return session
async def session_dependency(self) -> AsyncSession:
async def dispose(self) -> None:
await self.engine.dispose()
async def session_getter(self) -> AsyncGenerator[AsyncSession, None]:
async with self.session_factory() as session:
yield session
await session.close()
db_helper = DatabaseHelper(
url=settings.db.url,
url=str(settings.db.url),
echo=settings.db.echo,
echo_pool=settings.db.echo_pool,
pool_size=settings.db.pool_size,
max_overflow=settings.db.max_overflow,
)

View File

@ -1,7 +1,6 @@
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from .base import Base
from sqlalchemy.orm import Mapped, mapped_column
from core.models.base import Base
class Ip(Base):

View File

@ -16,5 +16,17 @@ class IpBase(BaseModel):
domain: str
class IpCreate(IpBase):
pass
class IpRead(IpBase):
id: int
pass
class IpUpdate(IpBase):
pass
class IpDelete(IpBase):
pass

27
main.py
View File

@ -3,13 +3,13 @@ from contextlib import asynccontextmanager
import uvicorn
from fastapi import FastAPI
from fastapi.openapi.docs import get_swagger_ui_html
from fastapi.responses import ORJSONResponse
from starlette.staticfiles import StaticFiles
from api import router as api_router
from core.config import BASE_DIR, settings
from core.models import db_helper
from core.config import STATIC_DIR, settings
from core.models.db_helper import db_helper
from api import router as swagger_router
log = logging.getLogger()
@ -26,24 +26,9 @@ main_app = FastAPI(
docs_url=None,
)
main_app.include_router(swagger_router)
main_app.mount(
"/static", StaticFiles(directory=BASE_DIR / "web" / "static"), name="static"
)
main_app.include_router(api_router)
@main_app.get("/docs", include_in_schema=False)
async def custom_swagger_ui_html():
return get_swagger_ui_html(
openapi_url=main_app.openapi_url,
title="Blocked IP API",
oauth2_redirect_url=main_app.swagger_ui_oauth2_redirect_url,
swagger_js_url="/static/swagger/swagger-ui-bundle.js",
swagger_css_url="/static/swagger/swagger-ui.css",
swagger_favicon_url="/static/swagger/favicon.png",
)
main_app.mount("/static", StaticFiles(directory=STATIC_DIR), name="static")
if __name__ == "__main__":
uvicorn.run(

View File

Before

Width:  |  Height:  |  Size: 4.9 KiB

After

Width:  |  Height:  |  Size: 4.9 KiB

5
utils/__init__.py Normal file
View File

@ -0,0 +1,5 @@
from utils.case_converter import camel_case_to_snake_case
__all__ = [
"camel_case_to_snake_case",
]

29
utils/case_converter.py Normal file
View File

@ -0,0 +1,29 @@
"""
Taken from
https://github.com/mahenzon/ri-sdk-python-wrapper/blob/master/ri_sdk_codegen/utils/case_converter.py
"""
def camel_case_to_snake_case(input_str: str) -> str:
"""
>>> camel_case_to_snake_case("SomeSDK")
'some_sdk'
>>> camel_case_to_snake_case("RServoDrive")
'r_servo_drive'
>>> camel_case_to_snake_case("SDKDemo")
'sdk_demo'
"""
chars = []
for c_idx, char in enumerate(input_str):
if c_idx and char.isupper():
nxt_idx = c_idx + 1
# idea of the flag is to separate abbreviations
# as new words, show them in lower case
flag = nxt_idx >= len(input_str) or input_str[nxt_idx].isupper()
prev_char = input_str[c_idx - 1]
if prev_char.isupper() and flag:
pass
else:
chars.append("_")
chars.append(char.lower())
return "".join(chars)