fixed incorrect typing issues

This commit is contained in:
2023-11-26 13:41:42 +01:00
parent 530b397592
commit 80ccaa83d3
8 changed files with 92 additions and 60 deletions

View File

@@ -1,6 +1,6 @@
from enum import Enum
from pydantic import BaseModel
from pydantic import BaseModel, Field
class Status(Enum):
@@ -12,3 +12,43 @@ class Status(Enum):
class Machine(BaseModel):
name: str
status: Status
class RepositoryBase(BaseModel):
title: str
description: str | None = None
class RepositoryCreate(RepositoryBase):
pass
class Repository(RepositoryBase):
id: int
prod_id: str
class Config:
orm_mode = True
class ProducerBase(BaseModel):
id: int
class ProducerCreate(ProducerBase):
jsonblob: int = Field(
42,
title="The Json",
description="this is the value of json",
gt=30,
lt=50,
list=[1, 2, "3"],
)
class Producer(ProducerBase):
id: int
repos: list[Repository] = []
class Config:
orm_mode = True

View File

@@ -8,4 +8,6 @@ router = APIRouter()
@router.get("/health", include_in_schema=True)
async def health() -> Machine: # str:
return Machine(name="test", status=Status.ONLINE)
# return "OK"

View File

@@ -1,16 +1,25 @@
from typing import List
from fastapi import APIRouter, Depends
from sqlalchemy.orm import Session
from .. import sql_crud, sql_db, sql_schemas
from .. import sql_crud, sql_db, sql_models
from ..api_outputs import Producer, ProducerCreate
router = APIRouter()
@router.get("/get_producers", response_model=list[sql_schemas.Producer])
def get_producers(skip: int = 0, limit: int = 100, db: Session = Depends(sql_db.get_db)):
@router.get("/get_producers", response_model=List[Producer])
def get_producers(
skip: int = 0, limit: int = 100, db: Session = Depends(sql_db.get_db)
) -> List[sql_models.Producer]:
producers = sql_crud.get_producers(db, skip=skip, limit=limit)
return producers
@router.post("/create_producers", response_model=sql_schemas.Producer)
def create_producers(producer: sql_schemas.ProducerCreate, db: Session = Depends(sql_db.get_db)):
@router.post("/create_producers", response_model=Producer)
def create_producers(
producer: ProducerCreate, db: Session = Depends(sql_db.get_db)
) -> Producer:
# todo checken ob schon da ...
return sql_crud.create_producer(db=db, producer=producer)

View File

@@ -1,12 +1,19 @@
from typing import List
from sqlalchemy.orm import Session
from . import sql_models, sql_schemas
from . import api_outputs, sql_models
def get_producers(db: Session, skip: int = 0, limit: int = 100):
def get_producers(
db: Session, skip: int = 0, limit: int = 100
) -> List[sql_models.Producer]:
return db.query(sql_models.Producer).offset(skip).limit(limit).all()
def create_producer(db: Session, producer: sql_schemas.ProducerCreate):
def create_producer(
db: Session, producer: api_outputs.ProducerCreate
) -> sql_models.Producer:
jsonblob_init = {"test_repo": "jsonblob_create"}
db_producer = sql_models.Producer(jsonblob=jsonblob_init)
db.add(db_producer)
@@ -15,10 +22,15 @@ def create_producer(db: Session, producer: sql_schemas.ProducerCreate):
return db_producer
def get_repositories(db: Session, skip: int = 0, limit: int = 100):
def get_repositories(
db: Session, skip: int = 0, limit: int = 100
) -> List[sql_models.Repository]:
return db.query(sql_models.Repository).offset(skip).limit(limit).all()
def create_repository(db: Session, repository: sql_schemas.RepositoryCreate, producers_id: int):
def create_repository(
db: Session, repository: api_outputs.RepositoryCreate, producers_id: int
) -> sql_models.Repository:
db_repository = sql_models.Repository(**repository.dict(), prod_id=producers_id)
db.add(db_repository)
db.commit()

View File

@@ -1,18 +1,19 @@
from typing import Generator
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import Session, sessionmaker
URL = "sqlite:///./sql_app.db"
engine = create_engine(
URL, connect_args={"check_same_thread":False}
)
engine = create_engine(URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
# Dependency to start a clean thread of the db
def get_db():
def get_db() -> Generator[Session, None, None]:
db = SessionLocal()
try:
yield db

View File

@@ -1,7 +1,9 @@
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, JSON
from sqlalchemy import JSON, Column, ForeignKey, Integer
from sqlalchemy.orm import relationship
from .sql_db import Base
class Producer(Base):
__tablename__ = "producers"
@@ -10,6 +12,7 @@ class Producer(Base):
repos = relationship("Repository", back_populates="producer")
class Repository(Base):
__tablename__ = "repositories"

View File

@@ -1,35 +0,0 @@
from pydantic import BaseModel, Field
class RepositoryBase(BaseModel):
title: str
description: str | None = None
class RepositoryCreate(RepositoryBase):
pass
class Repository(RepositoryBase):
id: int
prod_id: str
class Config:
orm_mode = True
class ProducerBase(BaseModel):
id: int
class ProducerCreate(ProducerBase):
jsonblob: int = Field(
42,
title='The Json',
description='this is the value of json',
gt=30,
lt=50,
list=[1,2,"3"],
)
class Producer(ProducerBase):
id: int
repos: list[Repository] = []
class Config:
orm_mode = True