config: use pydantic_settings; apitest: with logging, config improvements

This commit is contained in:
Helmut Merz 2026-09-20 08:01:26 +02:00
parent 5478ce98a7
commit f01e231e7a
6 changed files with 68 additions and 47 deletions

View file

@ -1,31 +1,44 @@
# py-scopes/apidemo/config.py # py-scopes/apidemo/config.py
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file='.env', extra='ignore')
#, env_prefix='SCOPES_')
base_url: str = 'https://test.cy7.eu'
log_file: str = 'log/scopes.log'
log_level: str = 'INFO'
log_format: str = '%(asctime)s %(levelname)s %(name)s %(message)s'
log_dateformat: str = '%Y-%m-%dT%H:%M:%S'
dbengine: str = 'postgresql+psycopg'
dbname: str = 'demo'
dbuser: str = 'demo'
dbpassword: str = 'secret'
dbschema: str = 'demo'
settings = Settings()
from dotenv import load_dotenv from dotenv import load_dotenv
import logging import logging
from os import getenv from os import getenv
from scopes.web.app import zope_app_factory
load_dotenv() load_dotenv()
log_file = 'log/scopes.log' base_url = settings.base_url
log_level = logging.DEBUG
log_format = '%(asctime)s %(levelname)s %(name)s %(message)s'
log_dateformat = '%Y-%m-%dT%H:%M:%S'
logging.basicConfig(filename=log_file, level=log_level,
format=log_format, datefmt=log_dateformat)
server_port = getenv('SERVER_PORT', '8099') logging.basicConfig(filename=settings.log_file, level=settings.log_level,
base_url = getenv('BASE_URL', 'https://demo.cy7.de') format=settings.log_format, datefmt=settings.log_dateformat)
logging.getLogger('uvicorn.error').propagate = True
app_factory = zope_app_factory
# storage settings # storage settings
from scopes.storage.db.postgres import StorageFactory from scopes.storage.db.postgres import StorageFactory
dbengine = 'postgresql+psycopg' dbengine = settings.dbengine
dbname = getenv('DBNAME', 'demo') dbname = settings.dbname
dbuser = getenv('DBUSER', 'demo') dbuser = settings.dbuser
dbpassword = getenv('DBPASSWORD', 'secret') dbpassword = settings.dbpassword
dbschema = getenv('DBSCHEMA', 'demo') dbschema = settings.dbschema
# authentication settings # authentication settings
oidc_provider = getenv('OIDC_PROVIDER', 'https://a1.cy7.de') oidc_provider = getenv('OIDC_PROVIDER', 'https://a1.cy7.de')

View file

@ -11,9 +11,7 @@ license = {text = "MIT"}
keywords = ["scopes"] keywords = ["scopes"]
authors = [{name = "Helmut Merz", email = "helmutm@cy55.de"}] authors = [{name = "Helmut Merz", email = "helmutm@cy55.de"}]
dependencies = [ dependencies = ["SQLAlchemy", "pydantic-settings"]
"SQLAlchemy",
]
[project.optional-dependencies] [project.optional-dependencies]
postgres = [ postgres = [
@ -31,7 +29,7 @@ app = [
] ]
auth = ["pyjwt[crypto]", "cryptography", "requests"] auth = ["pyjwt[crypto]", "cryptography", "requests"]
api = ["fastapi[standard]"] api = ["fastapi[standard]"]
test = ["zope.testrunner"] test = ["zope.testrunner", "pydantic-settings"]
apitest = ["pytest", "httpx2"] apitest = ["pytest", "httpx2"]
[tool.setuptools] [tool.setuptools]

View file

@ -1,5 +1,5 @@
# runtests.sh # runtests.sh
# run all unit / doc tests # run all unit / doc tests
zope-testrunner --test-path=. $* zope-testrunner --test-path=. -m !test_api $*

View file

@ -5,19 +5,20 @@ if '' not in sys.path:
sys.path = [''] + sys.path sys.path = [''] + sys.path
import config import config
config.storageFactory = config.StorageFactory(config)
from scopes.storage import tracking from scopes.storage import tracking
from scopes.storage.common import Storage from scopes.storage.common import Storage
import transaction import transaction
from fastapi import Depends, FastAPI from fastapi import Depends, FastAPI
from fastapi.responses import HTMLResponse
from typing import Annotated from typing import Annotated
# dependencies # dependencies
def storage(): def storage():
return config.storageFactory(config.dbschema) stf = config.StorageFactory(config)
return stf(config.dbschema)
# app and routes # app and routes
@ -27,13 +28,16 @@ app = FastAPI()
async def read_root(): async def read_root():
return {'Hello': 'World'} return {'Hello': 'World'}
@app.get('/demo')
async def read_demo():
return HTMLResponse('<b>Hello World</b>')
@app.get('/persons/{name}') @app.get('/persons/{name}')
async def read_person(name: str, q: str | None = None): async def read_person(name: str, q: str | None = None):
return {'name': name, 'query': q} return {'name': name, 'query': q}
@app.get('/tracks') @app.get('/tracks')
async def get_tracks(storage: Annotated[Storage, Depends(storage)]): async def get_tracks(storage: Annotated[Storage, Depends(storage)]):
#storage = config.storageFactory(config.dbschema)
tracks = storage.create(tracking.Container) tracks = storage.create(tracking.Container)
result = [tr.asDict() for tr in tracks.query()] result = [tr.asDict() for tr in tracks.query()]
transaction.commit() transaction.commit()

View file

@ -1,29 +1,33 @@
# py-scopes/tests/config.py # py-scopes/tests/config.py
import logging import logging
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
model_config = SettingsConfigDict(env_file='.env')
base_url: str = 'https://test.cy7.eu'
log_level: str = 'INFO'
dbengine: str = 'sqlite'
dbname: str = 'var/test.db'
dbuser: str | None = None
dbpassword: str | None = None
dbschema: str | None = None
settings = Settings()
from os import getenv from os import getenv
import sys import sys
#from scopes.web.app import demo_app, zope_app base_url = settings.base_url
log_file = 'scopes/tests/log/scopes-test.log' # SQLite storage settings
log_level = logging.INFO dbengine = settings.dbengine
log_format = '%(asctime)s %(levelname)s %(name)s %(message)s' dbname = settings.dbname
log_dateformat = '%Y-%m-%dT%H:%M:%S' dbuser = settings.dbuser
dbpassword = settings.dbpassword
# server / app settings dbschema = settings.dbschema
server_port = '8999'
base_url = 'testing:'
#app = zope_app
# storage settings
# SQLite
dbengine = 'sqlite'
dbname = 'var/test.db'
dbuser = None
dbpassword = None
dbschema = None
# special testing stuff # special testing stuff
from scopes.tests import data_auth # add oidc URIs and keys to dummy_requests data from scopes.tests import data_auth # add oidc URIs and keys to dummy_requests data

View file

@ -6,7 +6,7 @@ import os, sys
sys.path = [os.path.dirname(__file__)] + sys.path sys.path = [os.path.dirname(__file__)] + sys.path
import config import config
from scopes.storage.db.postgres import StorageFactory from scopes.storage.db import postgres
from scopes.storage import concept, folder, message, topic, tracking from scopes.storage import concept, folder, message, topic, tracking
import transaction import transaction
@ -15,7 +15,7 @@ config.dbname = 'testdb'
config.dbuser = 'testuser' config.dbuser = 'testuser'
config.dbpassword = 'secret' config.dbpassword = 'secret'
config.dbschema = 'testing' config.dbschema = 'testing'
config.storageFactory = StorageFactory(config) config.StorageFactory = postgres.StorageFactory
import pytest import pytest
from fastapi.testclient import TestClient from fastapi.testclient import TestClient
@ -26,7 +26,7 @@ from scopes.api.main import app
@pytest.fixture(scope='module') @pytest.fixture(scope='module')
def storage(): def storage():
#storage = config.storageFactory(config.dbschema) #storage = config.StorageFactory(config)(config.dbschema)
storage = main.storage() storage = main.storage()
storage.dropTable('tracks') storage.dropTable('tracks')
tracks = storage.create(tracking.Container) tracks = storage.create(tracking.Container)
@ -50,12 +50,14 @@ def test_root(storage):
assert response.status_code == 200 assert response.status_code == 200
assert response.json() == {'Hello': 'World'} assert response.json() == {'Hello': 'World'}
def test_tracking(storage): def test_tracking(storage, caplog):
with TestClient(app) as clt: with TestClient(app) as clt:
caplog.set_level(config.settings.log_level)
tracks = clt.get('/tracks').json() tracks = clt.get('/tracks').json()
assert len(tracks) == 1 assert len(tracks) == 1
tr = tracks[0] tr = tracks[0]
tr['timeStamp'] = '...' tr['timeStamp'] = '...'
assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'}, assert tr == {'uid': 'rec-1', 'head': {'taskId': 't01', 'userName': 'john'},
'data': {'activity': 'testing'}, 'timeStamp': '...'} 'data': {'activity': 'testing'}, 'timeStamp': '...'}
print('\nLogs: ', caplog.text)