This commit is contained in:
Waylon Walker 2023-05-19 08:35:16 -05:00
commit 4be274d9e2
No known key found for this signature in database
GPG key ID: 66E2BF2B4190EFE4
39 changed files with 2548 additions and 0 deletions

View file

@ -0,0 +1,4 @@
# SPDX-FileCopyrightText: 2023-present Waylon S. Walker <waylon@waylonwalker.com>
#
# SPDX-License-Identifier: MIT
__version__ = "0.0.0.dev1"

View file

@ -0,0 +1,3 @@
# SPDX-FileCopyrightText: 2023-present Waylon S. Walker <waylon@waylonwalker.com>
#
# SPDX-License-Identifier: MIT

View file

@ -0,0 +1,9 @@
# SPDX-FileCopyrightText: 2023-present Waylon S. Walker <waylon@waylonwalker.com>
#
# SPDX-License-Identifier: MIT
import sys
if __name__ == '__main__':
from .cli import {{python_package}}
sys.exit({{python_package}}())

64
learn_sql_model/api.py Normal file
View file

@ -0,0 +1,64 @@
from typing import Union
from fastapi import FastAPI
import httpx
from learn_sql_model.console import console
from learn_sql_model.models import Hero, Pet
models = Union[Hero, Pet]
# from learn_sql_model.config import config
# from learn_sql_model.models import Hero
app = FastAPI()
app.post("/heroes/")
def post(self: models) -> None:
try:
httpx.post("http://localhost:5000/heroes/", json=self.dict())
except httpx.ConnectError:
console.log("local failover")
post_local(self)
def post_local(self: models) -> None:
from learn_sql_model.config import config
with config.session as session:
session.add(self)
session.commit()
def get(self: models, instance: models = None) -> list[models]:
"read all the heros"
from learn_sql_model.config import config
with config.session as session:
if instance is None:
heroes = session.exec(select(self)).all()
return heroes
else:
hero = session.exec(select(self).where(self.id == instance.id)).all().one()
return hero
@app.post("/heroes/")
def create_hero(hero: Hero):
post(hero)
@app.get("/heroes/")
def read_heroes() -> list[Hero]:
"read all the heros"
return get(Hero)
@app.get("/hero/")
def read_heroes(hero: Hero) -> list[Hero]:
"read all the heros"
return get(Hero, hero)

View file

View file

@ -0,0 +1,9 @@
from fastapi import FastAPI
from learn_sql_model.api.hero import hero_router
from learn_sql_model.api.user import user_router
app = FastAPI()
app.include_router(hero_router)
app.include_router(user_router)

View file

@ -0,0 +1,31 @@
from typing import Annotated
from fastapi import APIRouter, Depends
from learn_sql_model.api.user import oauth2_scheme
from learn_sql_model.models import Hero
hero_router = APIRouter()
@hero_router.get("/items/")
async def read_items(token: Annotated[str, Depends(oauth2_scheme)]):
return {"token": token}
@hero_router.get("/hero/{id}")
def get_hero(id: int) -> Hero:
"get one hero"
return Hero.get(item_id=id)
@hero_router.post("/hero/")
def post_hero(hero: Hero) -> Hero:
"read all the heros"
return hero.post()
@hero_router.get("/heros/")
def get_heros() -> list[Hero]:
"get all heros"
return Hero.get()

146
learn_sql_model/api/user.py Normal file
View file

@ -0,0 +1,146 @@
from datetime import datetime, timedelta
from typing import Annotated
from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from pydantic import BaseModel
# to get a string like this run:
# openssl rand -hex 32
SECRET_KEY = "09d25e094faa6ca2556c818166b7a9563b93f7099f6f0f4caa6cf63b88e8d3e7"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
user_router = APIRouter()
fake_users_db = {
"johndoe": {
"username": "johndoe",
"full_name": "John Doe",
"email": "johndoe@example.com",
"hashed_password": "$2b$12$EixZaYVK1fsbw1ZfbX3OXePaWxn96p36WQoeG6Lruj3vjPGga31lW",
"disabled": False,
}
}
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
username: str | None = None
class User(BaseModel):
username: str
email: str | None = None
full_name: str | None = None
disabled: bool | None = None
class UserInDB(User):
hashed_password: str
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def get_user(db, username: str):
if username in db:
user_dict = db[username]
return UserInDB(**user_dict)
def authenticate_user(fake_db, username: str, password: str):
user = get_user(fake_db, username)
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
def create_access_token(data: dict, expires_delta: timedelta | None = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def get_current_user(token: Annotated[str, Depends(oauth2_scheme)]):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
username: str = payload.get("sub")
if username is None:
raise credentials_exception
token_data = TokenData(username=username)
except JWTError:
raise credentials_exception
user = get_user(fake_users_db, username=token_data.username)
if user is None:
raise credentials_exception
return user
async def get_current_active_user(
current_user: Annotated[User, Depends(get_current_user)]
):
if current_user.disabled:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
@user_router.post("/token", response_model=Token)
async def login_for_access_token(
form_data: Annotated[OAuth2PasswordRequestForm, Depends()]
):
user = authenticate_user(fake_users_db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect username or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.username}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@user_router.get("/users/me/", response_model=User)
async def read_users_me(
current_user: Annotated[User, Depends(get_current_active_user)]
):
return current_user
@user_router.get("/users/me/items/")
async def read_own_items(
current_user: Annotated[User, Depends(get_current_active_user)]
):
return [{"item_id": "Foo", "owner": current_user.username}]

View file

View file

@ -0,0 +1,28 @@
import typer
import uvicorn
from learn_sql_model.cli.common import verbose_callback
api_app = typer.Typer()
@api_app.callback()
def api(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
"model cli"
@api_app.command()
def run(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
uvicorn.run("learn_sql_model.api.app:app", port=5000, log_level="info")

View file

@ -0,0 +1,58 @@
import typer
from learn_sql_model.cli.api import api_app
from learn_sql_model.cli.common import verbose_callback
from learn_sql_model.cli.config import config_app
from learn_sql_model.cli.hero import hero_app
from learn_sql_model.cli.model_app import model_app
from learn_sql_model.cli.tui import tui_app
app = typer.Typer(
name="learn_sql_model",
help="A rich terminal report for coveragepy.",
)
app.add_typer(config_app)
app.add_typer(tui_app)
app.add_typer(model_app)
app.add_typer(api_app)
app.add_typer(hero_app, name="hero")
def version_callback(value: bool) -> None:
"""Callback function to print the version of the learn-sql-model package.
Args:
value (bool): Boolean value to determine if the version should be printed.
Raises:
typer.Exit: If the value is True, the version will be printed and the program will exit.
Example:
version_callback(True)
"""
if value:
from learn_sql_model.__about__ import __version__
typer.echo(f"{__version__}")
raise typer.Exit()
@app.callback()
def main(
version: bool = typer.Option(
None,
"--version",
callback=version_callback,
is_eager=True,
),
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
) -> None:
return
if __name__ == "__main__":
typer.run(main)

View file

@ -0,0 +1,6 @@
from learn_sql_model.console import console
def verbose_callback(value: bool) -> None:
if value:
console.quiet = False

View file

@ -0,0 +1,29 @@
from rich.console import Console
import typer
from learn_sql_model.cli.common import verbose_callback
from learn_sql_model.config import config as configuration
config_app = typer.Typer()
@config_app.callback()
def config(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
"configuration cli"
@config_app.command()
def show(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
Console().print(configuration)

View file

@ -0,0 +1,30 @@
from typing import List, Union
from pydantic_typer import expand_pydantic_args
from rich.console import Console
import typer
from learn_sql_model.models import Hero
hero_app = typer.Typer()
@hero_app.callback()
def hero():
"model cli"
@hero_app.command()
def get(id: int = None) -> Union[Hero, List[Hero]]:
"get one hero"
hero = Hero.get(item_id=id)
Console().print(hero)
return hero
@hero_app.command()
@expand_pydantic_args(typer=True)
def create(hero: Hero) -> Hero:
"read all the heros"
hero = hero.post()
Console().print(hero)

View file

@ -0,0 +1,119 @@
from rich.console import Console
from sqlmodel import SQLModel, Session
import typer
from learn_sql_model.cli.common import verbose_callback
from learn_sql_model.config import config
from learn_sql_model.models import Hero, Pet
model_app = typer.Typer()
@model_app.callback()
def model(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
"model cli"
@model_app.command()
def create_revision(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
message: str = typer.Option(
prompt=True,
),
):
import alembic
# python -m alembic revision --autogenerate -m "New Attribute"
from alembic.config import Config
alembic_cfg = Config("alembic.ini")
alembic.command.revision(
config=alembic_cfg,
message=message,
autogenerate=True,
)
@model_app.command()
def show(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
SQLModel.metadata.create_all(config.engine)
with Session(config.engine) as session:
heros = session.exec(select(Hero)).all()
Console().print(heros)
@model_app.command()
def read(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
from learn_sql_model.api import read_heroes
Console().print(read_heroes())
# @model_app.command()
# @expand_pydantic_args(typer=True)
# def create(
# hero: Hero,
# ):
# hero.post()
# try:
# httpx.post("http://localhost:5000/heroes/", json=hero.dict())
# except httpx.ConnectError:
# console.log("local failover")
# with Session(config.engine) as session:
# session.add(hero)
# session.commit()
@model_app.command()
def populate(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
pet_1 = Pet(name="Deadpond-Dog")
hero_1 = Hero(name="Deadpond", secret_name="Dive Wilson", pets=[pet_1])
hero_2 = Hero(
name="Spider-Boy",
secret_name="Pedro Parqueador",
pet=Pet(name="Spider-Boy-Dog"),
)
hero_3 = Hero(
name="Rusty-Man",
secret_name="Tommy Sharp",
age=48,
pet=Pet(name="Rusty-Man-Dog"),
)
SQLModel.metadata.create_all(config.engine)
with Session(config.engine) as session:
session.add(hero_1)
session.add(hero_2)
session.add(hero_3)
session.commit()

View file

@ -0,0 +1,18 @@
import typer
from learn_sql_model.cli.common import verbose_callback
from learn_sql_model.tui.app import run_app
tui_app = typer.Typer()
@tui_app.callback(invoke_without_command=True)
def i(
verbose: bool = typer.Option(
False,
callback=verbose_callback,
help="show the log messages",
),
):
"interactive tui"
run_app()

40
learn_sql_model/config.py Normal file
View file

@ -0,0 +1,40 @@
from typing import TYPE_CHECKING
from pydantic import BaseSettings
from sqlmodel import SQLModel, Session, create_engine
from learn_sql_model.models import Hero, Pet
from learn_sql_model.standard_config import load
models = [Hero, Pet]
if TYPE_CHECKING:
from sqlalchemy import Engine
class Config(BaseSettings):
database_url: str = "sqlite:///database.db"
class Config:
env_prefix = "LEARN_SQL_MODEL_"
@property
def engine(self) -> "Engine":
return create_engine(self.database_url)
@property
def session(self) -> "Session":
return Session(self.engine)
def create_db_and_tables(self) -> None:
SQLModel.metadata.create_all(self.engine)
# def create_endpoints(self) -> None:
# for model in models:
# app.post("/heroes/")(Hero.post_local)
# app.get("/heroes/")(Hero.read_heroes)
raw_config = load("learn_sql_model")
config = Config(**raw_config)
config.create_db_and_tables()

View file

@ -0,0 +1,4 @@
from rich.console import Console
console = Console()
console.quiet = False

43
learn_sql_model/models.py Normal file
View file

@ -0,0 +1,43 @@
from __future__ import annotations
from typing import Optional
from sqlmodel import Field, SQLModel, select
class FastModel(SQLModel):
def post(self):
from learn_sql_model.config import config
with config.session as session:
session.add(self)
session.commit()
@classmethod
def get(self, item_id: int = None):
from learn_sql_model.config import config
with config.session as session:
if item_id is None:
return session.exec(select(self)).all()
return session.exec(select(self).where(self.id == item_id)).one()
class Hero(FastModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str
secret_name: str
age: Optional[int] = None
# new_attribute: Optional[str] = None
# pets: List["Pet"] = Relationship(back_populates="hero")
class Pet(FastModel, table=True):
id: Optional[int] = Field(default=None, primary_key=True)
name: str = "Jim"
# age: Optional[int] = None
# hero_id: int = Field(default=None, foreign_key="hero.id")
# hero: Optional[Hero] = Relationship(back_populates="pets")

View file

@ -0,0 +1,239 @@
"""Standard Config.
A module to load tooling config from a users project space.
Inspired from frustrations that some tools have a tool.ini, .tool.ini,
setup.cfg, or pyproject.toml. Some allow for global configs, some don't. Some
properly follow the users home directory, others end up in a weird temp
directory. Windows home directory is only more confusing. Some will even
respect the users `$XDG_HOME` directory.
This file is for any project that can be configured in plain text such as `ini`
or `toml` and not requiring a .py file. Just name your tool and let users put
config where it makes sense to them, no need to figure out resolution order.
## Usage:
``` python
from standard_config import load
# Retrieve any overrides from the user
overrides = {'setting': True}
config = load('my_tool', overrides)
```
## Resolution Order
* First global file with a tool key
* First local file with a tool key
* Environment variables prefixed with `TOOL`
* Overrides
### Tool Specific Ini files
Ini file formats must include a `<tool>` key.
``` ini
[my_tool]
setting = True
```
### pyproject.toml
Toml files must include a `tool.<tool>` key
``` toml
[tool.my_tool]
setting = True
```
### setup.cfg
setup.cfg files must include a `tool:<tool>` key
``` ini
[tool:my_tool]
setting = True
```
### global files to consider
* <home>/tool.ini
* <home>/.tool
* <home>/.tool.ini
* <home>/.config/tool.ini
* <home>/.config/.tool
* <home>/.config/.tool.ini
### local files to consider
* <project_home>/tool.ini
* <project_home>/.tool
* <project_home>/.tool.ini
* <project_home>/pyproject.toml
* <project_home>/setup.cfg
"""
import os
from pathlib import Path
from typing import Dict, List, Union
import anyconfig
# path_spec_type = List[Dict[str, Union[Path, str, List[str\}\}\}\}
path_spec_type = List
def _get_global_path_specs(tool: str) -> path_spec_type:
"""
Generate a list of standard pathspecs for global config files.
Args:
tool (str): name of the tool to configure
"""
try:
home = Path(os.environ["XDG_HOME"])
except KeyError:
home = Path.home()
return [
{"path_specs": home / f"{tool}.ini", "ac_parser": "ini", "keys": [tool]},
{"path_specs": home / f".{tool}", "ac_parser": "ini", "keys": [tool]},
{"path_specs": home / f".{tool}.ini", "ac_parser": "ini", "keys": [tool]},
{
"path_specs": home / ".config" / f"{tool}.ini",
"ac_parser": "ini",
"keys": [tool],
},
{
"path_specs": home / ".config" / f".{tool}",
"ac_parser": "ini",
"keys": [tool],
},
{
"path_specs": home / ".config" / f".{tool}.ini",
"ac_parser": "ini",
"keys": [tool],
},
]
def _get_local_path_specs(tool: str, project_home: Union[str, Path]) -> path_spec_type:
"""
Generate a list of standard pathspecs for local, project directory config files.
Args:
tool (str): name of the tool to configure
"""
return [
{
"path_specs": Path(project_home) / f"{tool}.ini",
"ac_parser": "ini",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f".{tool}",
"ac_parser": "ini",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f".{tool}.ini",
"ac_parser": "ini",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f"{tool}.yml",
"ac_parser": "yaml",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f".{tool}.yml",
"ac_parser": "yaml",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f"{tool}.toml",
"ac_parser": "toml",
"keys": [tool],
},
{
"path_specs": Path(project_home) / f".{tool}.toml",
"ac_parser": "toml",
"keys": [tool],
},
{
"path_specs": Path(project_home) / "pyproject.toml",
"ac_parser": "toml",
"keys": ["tool", tool],
},
{
"path_specs": Path(project_home) / "setup.cfg",
"ac_parser": "ini",
"keys": [f"tool.{tool}"],
},
]
def _get_attrs(attrs: list, config: Dict) -> Dict:
"""Get nested config data from a list of keys.
specifically written for pyproject.toml which needs to get `tool` then `<tool>`
"""
for attr in attrs:
config = config[attr]
return config
def _load_files(config_path_specs: path_spec_type) -> Dict:
"""Use anyconfig to load config files stopping at the first one that exists.
config_path_specs (list): a list of pathspecs and keys to load
"""
for file in config_path_specs:
if file["path_specs"].exists():
config = anyconfig.load(**file)
else:
# ignore missing files
continue
try:
return _get_attrs(file["keys"], config)
except KeyError:
# ignore incorrect keys
continue
return {}
def _load_env(tool: str) -> Dict:
"""Load config from environment variables.
Args:
tool (str): name of the tool to configure
"""
vars = [var for var in os.environ.keys() if var.startswith(tool.upper())]
return {
var.lower().strip(tool.lower()).strip("_").strip("-"): os.environ[var]
for var in vars
}
def load(tool: str, project_home: Union[Path, str] = ".", overrides: Dict = {}) -> Dict:
"""Load tool config from standard config files.
Resolution Order
* First global file with a tool key
* First local file with a tool key
* Environment variables prefixed with `TOOL`
* Overrides
Args:
tool (str): name of the tool to configure
"""
global_config = _load_files(_get_global_path_specs(tool))
local_config = _load_files(_get_local_path_specs(tool, project_home))
env_config = _load_env(tool)
return {**global_config, **local_config, **env_config, **overrides}

View file

@ -0,0 +1,18 @@
Screen {
align: center middle;
layers: main footer;
}
Sidebar {
height: 100vh;
width: auto;
min-width: 20;
background: $secondary-background-darken-2;
dock: left;
margin-right: 1;
layer: main;
}
Footer {
layer: footer;
}

View file

@ -0,0 +1,56 @@
# config["tui"] = {}
# config["tui"]["bindings"] = {}
# class Sidebar(Static):
# def compose(self) -> ComposeResult:
# yield Container(
# Static("sidebar"),
# id="sidebar",
# )
# class Tui(App):
# """A Textual app to manage requests."""
# CSS_PATH = Path("__file__").parent / "app.css"
# BINDINGS = [tuple(b.values()) for b in config["tui"]["bindings"]]
# def compose(self) -> ComposeResult:
# """Create child widgets for the app."""
# yield Container(Static("hello world"))
# yield Footer()
# def action_toggle_dark(self) -> None:
# """An action to toggle dark mode."""
# self.dark = not self.dark
# def action_toggle_sidebar(self):
# try:
# self.query_one("PromptSidebar").remove()
# except NoMatches:
# self.mount(Sidebar())
def run_app():
...
# import os
# import sys
# from textual.features import parse_features
# dev = "--dev" in sys.argv
# features = set(parse_features(os.environ.get("TEXTUAL", "")))
# if dev:
# features.add("debug")
# features.add("devtools")
# os.environ["TEXTUAL"] = ",".join(sorted(features))
# app = Tui()
# app.run()
# if __name__ == "__main__":
# run_app()