129 lines
3.2 KiB
Python
129 lines
3.2 KiB
Python
from sqlalchemy.orm.attributes import InstrumentedAttribute
|
|
from sqlmodel import Session, SQLModel, create_engine, select
|
|
|
|
from app.config import settings
|
|
|
|
engine = create_engine(
|
|
settings.database_url,
|
|
echo=True,
|
|
connect_args={"check_same_thread": False},
|
|
)
|
|
|
|
|
|
def create_database():
|
|
SQLModel.metadata.create_all(engine)
|
|
|
|
|
|
### POST route
|
|
def add_to_database(input: SQLModel):
|
|
with Session(engine) as session:
|
|
session.add(input)
|
|
session.commit()
|
|
|
|
|
|
### GET "/all" route
|
|
def get_from_database[T: SQLModel](query: type[T]) -> list[T] | None:
|
|
with Session(engine) as session:
|
|
results = list(session.exec(select(query)))
|
|
return results
|
|
|
|
|
|
### GET "/{value_of_var}" route
|
|
def get_from_database_where[T: SQLModel, V](
|
|
query: type[T],
|
|
variable: InstrumentedAttribute[V],
|
|
value: V,
|
|
offset: int | None = None,
|
|
limit: int | None = None,
|
|
) -> list[T] | None:
|
|
with Session(engine) as session:
|
|
statement = select(query).where(variable == value)
|
|
if offset is not None:
|
|
statement = statement.offset(offset)
|
|
if limit is not None:
|
|
statement = statement.limit(limit)
|
|
results = list(session.exec(statement))
|
|
return results
|
|
|
|
|
|
### GET "/{id}" route
|
|
def get_from_database_where_id[T: SQLModel](
|
|
query: type[T],
|
|
value: int,
|
|
) -> T | None:
|
|
with Session(engine) as session:
|
|
result = session.get(query, value)
|
|
return result
|
|
|
|
|
|
### GET "/{value where x for y and y has link to z}" route
|
|
def get_from_database_where_join_filter[
|
|
T: SQLModel,
|
|
K: SQLModel,
|
|
V,
|
|
](
|
|
query: type[T],
|
|
joined: type[K],
|
|
variable: InstrumentedAttribute[V],
|
|
value: V,
|
|
) -> list[T] | None:
|
|
with Session(engine) as session:
|
|
results = list(
|
|
session.exec(select(query).join(joined).where(variable == value))
|
|
)
|
|
return results
|
|
|
|
|
|
### GET "/{with table y}" route (can have outers or not)
|
|
def get_and_join_from_database[
|
|
T: SQLModel,
|
|
K: SQLModel,
|
|
](
|
|
query: type[T],
|
|
joined: type[K],
|
|
isouter: bool = False,
|
|
) -> list[tuple[T, K]] | None:
|
|
with Session(engine) as session:
|
|
results = list(
|
|
session.exec(select(query, joined).join(joined, isouter=isouter))
|
|
)
|
|
return results
|
|
|
|
|
|
### PATCH "/{value_of_var}" route
|
|
def update_database_where[T: SQLModel, V, W](
|
|
query: type[T],
|
|
variable: InstrumentedAttribute[V],
|
|
value: V,
|
|
attribute: InstrumentedAttribute[W],
|
|
new_value: W,
|
|
) -> list[T]:
|
|
with Session(engine) as session:
|
|
results = list(session.exec(select(query).where(variable == value)))
|
|
|
|
for instance in results:
|
|
setattr(instance, attribute.key, new_value)
|
|
session.add(instance)
|
|
|
|
session.commit()
|
|
|
|
for instance in results:
|
|
session.refresh(instance)
|
|
return results
|
|
|
|
|
|
### DELETE "/{value_of_var}" route
|
|
def delete_from_database_where[T: SQLModel, V](
|
|
query: type[T],
|
|
variable: InstrumentedAttribute[V],
|
|
value: V,
|
|
) -> list[T]:
|
|
with Session(engine) as session:
|
|
results = list(session.exec(select(query).where(variable == value)))
|
|
|
|
for instance in results:
|
|
session.delete(instance)
|
|
|
|
session.commit()
|
|
|
|
return results
|