пробую переползти на fastui

This commit is contained in:
sergey 2024-07-13 20:56:34 +03:00
parent 8cf56bdc89
commit a4c7c00254
35 changed files with 659 additions and 150 deletions

View File

@ -7,8 +7,8 @@ from sqlalchemy.ext.asyncio import async_engine_from_config
from alembic import context
from core.config import settings
from core.models import Base
from settings import settings
from models import Base
# this is the Alembic Config object, which provides
# access to the values within the .ini file in use.
@ -29,7 +29,8 @@ target_metadata = Base.metadata
# can be acquired:
# my_important_option = config.get_main_option("my_important_option")
# ... etc.
config.set_main_option('sqlalchemy.url', str(settings.db.url))
config.set_main_option("sqlalchemy.url", str(settings.db.url))
def run_migrations_offline() -> None:
"""Run migrations in 'offline' mode.

View File

@ -0,0 +1,41 @@
"""create user table
Revision ID: b068e8be0d4b
Revises: 443e39c236a6
Create Date: 2024-07-13 13:27:25.512678
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "b068e8be0d4b"
down_revision: Union[str, None] = "443e39c236a6"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_constraint("uq_users_foo_bar", "users", type_="unique")
op.drop_column("users", "foo")
op.drop_column("users", "bar")
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.add_column(
"users",
sa.Column("bar", sa.INTEGER(), autoincrement=False, nullable=False),
)
op.add_column(
"users",
sa.Column("foo", sa.INTEGER(), autoincrement=False, nullable=False),
)
op.create_unique_constraint("uq_users_foo_bar", "users", ["foo", "bar"])
# ### end Alembic commands ###

View File

@ -0,0 +1,42 @@
"""create isp_connection table
Revision ID: 1855b4dcf566
Revises: b068e8be0d4b
Create Date: 2024-07-13 15:27:44.137123
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "1855b4dcf566"
down_revision: Union[str, None] = "b068e8be0d4b"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"isps",
sa.Column("name", sa.String(), nullable=False),
sa.Column("manager_name", sa.String(), nullable=True),
sa.Column("manager_phone", sa.String(), nullable=True),
sa.Column("manager_email", sa.String(), nullable=True),
sa.Column("tech_support_phone", sa.String(), nullable=True),
sa.Column("tesh_support_email", sa.String(), nullable=True),
sa.Column("comment", sa.String(), nullable=True),
sa.Column("id", sa.Integer(), nullable=False),
sa.PrimaryKeyConstraint("id", name=op.f("pk_isps")),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("isps")
# ### end Alembic commands ###

View File

@ -0,0 +1,51 @@
"""create isp_connection table
Revision ID: 85fef76b3dcd
Revises: 1855b4dcf566
Create Date: 2024-07-13 15:30:30.938434
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
# revision identifiers, used by Alembic.
revision: str = "85fef76b3dcd"
down_revision: Union[str, None] = "1855b4dcf566"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.create_table(
"isp_connections",
sa.Column("isp_name", sa.String(), nullable=False),
sa.Column("location_code", sa.String(), nullable=False),
sa.Column("contract_num", sa.String(), nullable=True),
sa.Column("contract_date", sa.String(), nullable=True),
sa.Column("contract_company", sa.String(), nullable=True),
sa.Column("cost", sa.Integer(), nullable=True),
sa.Column("speed", sa.Integer(), nullable=True),
sa.Column("connection_type", sa.String(), nullable=True),
sa.Column("network", sa.String(), nullable=True),
sa.Column("address_type", sa.String(), nullable=True),
sa.Column("isp_id", sa.Integer(), nullable=False),
sa.Column("id", sa.Integer(), nullable=False),
sa.ForeignKeyConstraint(
["isp_id"],
["isps.id"],
name=op.f("fk_isp_connections_isp_id_isps"),
),
sa.PrimaryKeyConstraint("id", name=op.f("pk_isp_connections")),
)
# ### end Alembic commands ###
def downgrade() -> None:
# ### commands auto generated by Alembic - please adjust! ###
op.drop_table("isp_connections")
# ### end Alembic commands ###

View File

@ -1,11 +1,23 @@
from fastapi import APIRouter
from core.config import settings
from .api_v1 import router as router_api_v1
from settings import settings
router = APIRouter(
prefix=settings.api.prefix
from .users import router as user_router
from .isp import router as isp_router
from .isp_connection import router as isp_connection_router
router = APIRouter()
router.include_router(
user_router,
prefix=settings.api.users,
)
router.include_router(
router_api_v1,
isp_router,
prefix=settings.api.isp,
)
router.include_router(
isp_connection_router,
prefix=settings.api.isp_connections,
)

View File

@ -1,13 +0,0 @@
from fastapi import APIRouter
from core.config import settings
from .users import router as users_router
router = APIRouter(
prefix=settings.api.v1.prefix,
)
router.include_router(users_router,
prefix=settings.api.v1.users,
)

View File

@ -1,27 +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.user import UserRead, UserCreate
from api.api_v1.crud import users as users_crud
router = APIRouter(
tags=['Users'],
)
@router.get('', response_model=list[UserRead])
async def get_users(session: Annotated[AsyncSession, Depends(db_helper.session_getter)]):
users = await users_crud.get_all_users(session=session)
return users
@router.post('', response_model=UserRead)
async def create_user(session: Annotated[AsyncSession, Depends(db_helper.session_getter)], user_create: UserCreate,):
user = await users_crud.create_user(session=session, user_create=user_create)
return user

28
sipi-app/api/isp.py Normal file
View File

@ -0,0 +1,28 @@
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from models import db_helper
from schemas.isp import IspRead, IspCreate
from crud import isp as isp_crud
router = APIRouter(
tags=["Isp"],
)
@router.get("", response_model=list[IspRead])
async def get_isp(session: Annotated[AsyncSession, Depends(db_helper.session_getter)]):
users = await isp_crud.get_all_isp(session=session)
return users
@router.post("", response_model=IspRead)
async def create_isp(
session: Annotated[AsyncSession, Depends(db_helper.session_getter)],
isp_scheme: IspCreate,
):
isp = await isp_crud.create_isp(session=session, isp_scheme=isp_scheme)
return isp

View File

@ -0,0 +1,32 @@
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from models import db_helper
from schemas.isp_connections import IspConnectionRead, IspConnectionCreate
from crud import isp_connection as isp_connections_crud
router = APIRouter(
tags=["IspConnections"],
)
@router.get("", response_model=list[IspConnectionRead])
async def get_isp_connection(
session: Annotated[AsyncSession, Depends(db_helper.session_getter)]
):
users = await isp_connections_crud.get_all_isp_connection(session=session)
return users
@router.post("", response_model=IspConnectionRead)
async def create_isp_connection(
session: Annotated[AsyncSession, Depends(db_helper.session_getter)],
isp_connection_create: IspConnectionCreate,
):
isp_connection = await isp_connections_crud.create_isp_connection(
session=session, isp_connection_scheme=isp_connection_create
)
return isp_connection

31
sipi-app/api/users.py Normal file
View File

@ -0,0 +1,31 @@
from typing import Annotated
from fastapi import APIRouter, Depends
from sqlalchemy.ext.asyncio import AsyncSession
from models import db_helper
from schemas.user import UserRead, UserCreate
from crud import user as user_crud
router = APIRouter(
tags=["User"],
)
@router.get("", response_model=list[UserRead])
async def get_users(
session: Annotated[AsyncSession, Depends(db_helper.session_getter)]
):
users = await user_crud.get_all_users(session=session)
return users
@router.post("", response_model=UserRead)
async def create_user(
session: Annotated[AsyncSession, Depends(db_helper.session_getter)],
user_create: UserCreate,
):
user = await user_crud.create_user(session=session, user_create=user_create)
return user

View File

@ -1,55 +0,0 @@
from pydantic import BaseModel
from pydantic import PostgresDsn
from pydantic import MariaDBDsn
from pydantic import MySQLDsn
from pydantic_settings import (
BaseSettings,
SettingsConfigDict,
)
class RunConfig(BaseModel):
host: str = '0.0.0.0'
port: int = 8000
reload: bool = True
class ApiV1Prefix(BaseModel):
prefix: str = '/v1'
users: str = '/users'
class ApiPrefix(BaseModel):
prefix: str = '/api'
v1: ApiV1Prefix = ApiV1Prefix()
class DatabaseConfig(BaseModel):
url: PostgresDsn
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_N_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 Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=('D:\PythonScripts\sipi-web\sipi-app\.env', 'sipi-app/.env'),
case_sensitive=False,
env_nested_delimiter='__',
env_prefix='SIPI_CONFIG__',
)
run: RunConfig = RunConfig()
api: ApiPrefix = ApiPrefix()
db: DatabaseConfig
settings = Settings()

View File

@ -1,9 +0,0 @@
__all__ = (
'db_helper',
'Base',
'User',
)
from .db_helper import db_helper
from .base import Base
from .user import User

20
sipi-app/crud/isp.py Normal file
View File

@ -0,0 +1,20 @@
from typing import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import Isp
from schemas.isp import IspCreate
async def get_all_isp(session: AsyncSession) -> Sequence[Isp]:
stmt = select(Isp).order_by(Isp.id)
result = await session.scalars(stmt)
return result.all()
async def create_isp(session: AsyncSession, isp_scheme: IspCreate) -> Isp:
isp = Isp(**isp_scheme.model_dump())
session.add(isp)
await session.commit()
return isp

View File

@ -0,0 +1,23 @@
from typing import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from models import IspConnection
from schemas.isp_connections import IspConnectionCreate
async def get_all_isp_connection(session: AsyncSession) -> Sequence[IspConnection]:
stmt = select(IspConnection).order_by(IspConnection.id)
result = await session.scalars(stmt)
return result.all()
async def create_isp_connection(
session: AsyncSession, isp_connection_scheme: IspConnectionCreate
) -> IspConnection:
isp_connection = IspConnection(**isp_connection_scheme.model_dump())
session.add(isp_connection)
await session.commit()
return isp_connection

View File

@ -3,8 +3,8 @@ from typing import Sequence
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from core.models import User
from core.schemas.user import UserCreate
from models import User
from schemas.user import UserCreate
async def get_all_users(session: AsyncSession) -> Sequence[User]:

View File

@ -1,32 +1,39 @@
from contextlib import asynccontextmanager
from core.config import settings
from core.models import db_helper
from starlette.staticfiles import StaticFiles
from settings import settings
from models import db_helper
from api import router as api_router
from views import router as web_router
import uvicorn
from fastapi import FastAPI
from fastapi.responses import ORJSONResponse
@asynccontextmanager
async def lifespan(app: FastAPI):
yield
await db_helper.dispose()
main_app = FastAPI(
default_response_class=ORJSONResponse,
lifespan=lifespan,
)
main_app.include_router(api_router,
prefix=settings.api.prefix)
main_app.include_router(api_router)
main_app.include_router(web_router)
main_app.mount("/static", StaticFiles(directory="views/static"), name="static")
if __name__ == '__main__':
uvicorn.run('main:main_app',
if __name__ == "__main__":
uvicorn.run(
"main:main_app",
host=settings.run.host,
port=settings.run.port,
reload=settings.run.reload)
reload=settings.run.reload,
)

View File

@ -0,0 +1,13 @@
__all__ = (
"db_helper",
"Base",
"User",
"Isp",
"IspConnection",
)
from .db_helper import db_helper
from .base import Base
from .user import User
from .isp import Isp
from .isp_connection import IspConnection

View File

@ -1,19 +1,19 @@
from sqlalchemy import MetaData
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, declared_attr
from core.config import settings
from settings import settings
from utils import camel_case_to_snake_case
class Base(DeclarativeBase):
__abstract__ = True
metadata = MetaData(
naming_convention=settings.db.naming_convention,
)
@declared_attr.directive
def __tablename__(cls) -> str:
return f'{camel_case_to_snake_case(cls.__name__)}s'
return f"{camel_case_to_snake_case(cls.__name__)}s"
id: Mapped[int] = mapped_column(primary_key=True)

View File

@ -5,11 +5,12 @@ from sqlalchemy.ext.asyncio import AsyncSession
from typing import AsyncGenerator
from core.config import settings
from settings import settings
class DatabaseHelper:
def __init__(self,
def __init__(
self,
url: str,
echo: bool = False,
echo_pool: bool = False,

21
sipi-app/models/isp.py Normal file
View File

@ -0,0 +1,21 @@
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from .base import Base
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from isp_connection import IspConnection
class Isp(Base):
name: Mapped[str]
manager_name: Mapped[str] = mapped_column(nullable=True)
manager_phone: Mapped[str] = mapped_column(nullable=True)
manager_email: Mapped[str] = mapped_column(nullable=True)
tech_support_phone: Mapped[str] = mapped_column(nullable=True)
tesh_support_email: Mapped[str] = mapped_column(nullable=True)
comment: Mapped[str] = mapped_column(nullable=True)
isp_connections: Mapped[list["IspConnection"]] = relationship(back_populates="isp")

View File

@ -0,0 +1,23 @@
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy import ForeignKey
from .base import Base
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from .isp import Isp
class IspConnection(Base):
location_code: Mapped[str]
contract_num: Mapped[str] = mapped_column(nullable=True)
contract_date: Mapped[str] = mapped_column(nullable=True)
contract_company: Mapped[str] = mapped_column(nullable=True)
cost: Mapped[int] = mapped_column(nullable=True)
speed: Mapped[int] = mapped_column(nullable=True)
connection_type: Mapped[str] = mapped_column(nullable=True)
network: Mapped[str] = mapped_column(nullable=True)
address_type: Mapped[str] = mapped_column(nullable=True)
isp_id: Mapped[int] = mapped_column(ForeignKey("isps.id"))
isp: Mapped["Isp"] = relationship(back_populates="isp_connections")

View File

@ -1,4 +1,3 @@
from sqlalchemy import UniqueConstraint
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from .base import Base
@ -6,9 +5,3 @@ from .base import Base
class User(Base):
username: Mapped[str] = mapped_column(unique=True)
foo: Mapped[int]
bar: Mapped[int]
__table_args__ = (
UniqueConstraint('foo', 'bar'),
)

27
sipi-app/schemas/isp.py Normal file
View File

@ -0,0 +1,27 @@
from pydantic import BaseModel
class IspBase(BaseModel):
name: str
manager_name: str
manager_phone: str
manager_email: str
tech_support_phone: str
tesh_support_email: str
comment: str
class IspCreate(IspBase):
pass
class IspRead(IspBase):
id: int
class IspRemove(IspBase):
pass
class IspChange(IspBase):
pass

View File

@ -0,0 +1,29 @@
from pydantic import BaseModel
class IspConnectionBase(BaseModel):
location_code: str
contract_num: str
contract_date: str
contract_company: str
cost: int
speed: int
connection_type: str
network: str
address_type: str
class IspConnectionCreate(IspConnectionBase):
isp_id: int
class IspConnectionRead(IspConnectionBase):
id: int
class IspConnectionRemove(IspConnectionBase):
pass
class IspConnectionChange(IspConnectionBase):
pass

View File

@ -3,8 +3,6 @@ from pydantic import BaseModel
class UserBase(BaseModel):
username: str
foo: int
bar: int
class UserCreate(UserBase):
@ -21,5 +19,3 @@ class UserRemove(UserBase):
class UserChange(UserBase):
pass

59
sipi-app/settings.py Normal file
View File

@ -0,0 +1,59 @@
from pydantic import BaseModel
from pydantic import PostgresDsn
# from pydantic import MariaDBDsn
# from pydantic import MySQLDsn
from pydantic_settings import (
BaseSettings,
SettingsConfigDict,
)
class RunConfig(BaseModel):
host: str = "0.0.0.0"
port: int = 8000
reload: bool = True
class ApiPrefix(BaseModel):
users: str = "/api/users"
isp: str = "/api/isp"
isp_connections: str = "/api/isp_connections"
class WebPrefix(BaseModel):
start: str = "/web"
isp: str = "/web/isp"
isp_connections: str = "/web/isp_connections"
class DatabaseConfig(BaseModel):
url: PostgresDsn
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_N_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 Settings(BaseSettings):
model_config = SettingsConfigDict(
env_file=(".env-template", ".env"),
case_sensitive=False,
env_nested_delimiter="__",
env_prefix="SIPI_CONFIG__",
)
run: RunConfig = RunConfig()
api: ApiPrefix = ApiPrefix()
web: WebPrefix = WebPrefix()
db: DatabaseConfig
settings = Settings()

View File

@ -0,0 +1,17 @@
from fastapi import APIRouter
from settings import settings
from .web import router as web_router
from .isp import router as isp_router
router = APIRouter()
router.include_router(
web_router,
prefix=settings.web.start,
)
router.include_router(
isp_router,
prefix=settings.web.isp,
)

23
sipi-app/views/isp.py Normal file
View File

@ -0,0 +1,23 @@
from fastapi import APIRouter
from starlette.requests import Request
from starlette.responses import HTMLResponse
from starlette.templating import Jinja2Templates
from settings import settings
router = APIRouter(
tags=["Web_Isp"],
)
templates = Jinja2Templates(directory="views/templates")
@router.get("/", response_class=HTMLResponse)
async def read_item(request: Request):
print("sadf")
return templates.TemplateResponse(
"index.html",
{
"request": request,
},
)

View File

@ -0,0 +1,35 @@
h1 {
color: green;
}
button {
background-color: gray;
width: 250px;
color: white;
text-align: left;
}
ping.button.button {
background-color: green;
width: 250px;
color: white;
text-align: left;
}
td {
word-wrap:break-word;
border: 1px solid grey;
}
table {
display: inline;
}
.htmx-indicator{
display:none;
}
.htmx-request .htmx-indicator{
display:inline;
}
.htmx-request.htmx-indicator{
display:inline;
}

View File

@ -0,0 +1,25 @@
<html>
<head>
<title>Hello...</title>
</head>
<body>
<div id="parent-div">
<h1>Hello...</h1>
</div>
<div id="tabs" hx-target="#tab-contents" role="tablist" _="on htmx:afterOnLoad set @aria-selected of <[aria-selected=true]/> to false tell the target take .selected set @aria-selected to true">
<button role="tab" aria-controls="tab-content" aria-selected="true" hx-get="/tab1" class="selected">Tab 1</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab2">Tab 2</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab3">Tab 3</button>
</div>
<div id="tab-contents" role="tabpanel" hx-get="/tab1" hx-trigger="load"></div>
<button
hx-get="/world"
hx-trigger="click"
hx-target="#parent-div"
hx-swap="innerHTML"
>Click Me!
</button>
<script src="/static/js/htmx.js"></script>
</body>
</html>

View File

@ -0,0 +1,31 @@
<!DOCTYPE HTML>
<html>
<head>
<title>SiPi-web</title>
<link href="/static/css/isp.css" rel="stylesheet" type="text/css" />
<link rel="icon" href="data:;base64,=">
<link rel="stylesheet" href="https://cdnjs.cloudflare.com/ajax/libs/font-awesome/4.7.0/css/font-awesome.min.css">
<script src="/static/js/htmx.js"></script>
</head>
<div id="tabs" hx-target="#tab-contents" role="tablist" _="on htmx:afterOnLoad set @aria-selected of <[aria-selected=true]/> to false tell the target take .selected set @aria-selected to true">
<button role="tab" aria-controls="tab-content" aria-selected="true" hx-get="/tab1" class="selected">Площадки</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab2">Провайдеры</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab3">Подключения</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab5">Оборудование</button>
<button role="tab" aria-controls="tab-content" aria-selected="false" hx-get="/tab5">Подсети</button>
</div>
<div id="tab-contents" role="tabpanel" hx-get="/tab1" hx-trigger="load">
<h1>Провайдеры</h1>
</div>
<div id="tab-contents" role="tabpanel" hx-get="/tab2" hx-trigger="load">
<h1>Провайдеры</h1>
</div>
<body>
</body>
</html>

32
sipi-app/views/web.py Normal file
View File

@ -0,0 +1,32 @@
from fastapi import APIRouter
from starlette.requests import Request
from starlette.responses import HTMLResponse
from starlette.templating import Jinja2Templates
from settings import settings
router = APIRouter(
tags=["Web_Start"],
)
templates = Jinja2Templates(directory="views/templates")
@router.get("/", response_class=HTMLResponse)
async def read_item(request: Request):
return templates.TemplateResponse(
"index.html",
{
"request": request,
},
)
@router.get("/", response_class=HTMLResponse)
async def read_item(request: Request):
return templates.TemplateResponse(
"index.html",
{
"request": request,
},
)