API reference¶
This package is a CLI plugin rather than a library, so it declares no __all__. The modules below are its entry points.
auth_module_config¶
Configuração dos módulos RBAC Auth.
module-attribute
¶
Python
RBAC_MODULES: dict[str, AuthModuleDef] = {'core': AuthModuleDef(name='core', entities=[EntityDef(name='auth_user', fields_str='email:str,username:str,password_hash:str,first_name:str?,last_name:str?,is_active:bool=True,is_superuser:bool=False,status:str=active,tenant_id:str?', enums=['status:active|inactive|locked|pending']), EntityDef(name='role', fields_str='name:str,description:str?,is_system:bool=False,tenant_id:str?'), EntityDef(name='permission', fields_str='name:str,description:str?,resource:str,action:str,tenant_id:str?'), EntityDef(name='session', fields_str='user_id:str,token:str,ip_address:str?,user_agent:str?,expires_at:str,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='refresh_token', fields_str='user_id:str,token:str,expires_at:str,is_revoked:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='login_attempt', fields_str='user_id:str?,email:str,ip_address:str?,success:bool,failure_reason:str?,tenant_id:str?'), EntityDef(name='password_reset', fields_str='user_id:str,token:str,expires_at:str,is_used:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='email_verification', fields_str='user_id:str,token:str,expires_at:str,is_verified:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='login_code', fields_str='user_id:str,code:str,expires_at:str,is_used:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='token_blacklist', fields_str='token:str,blacklisted_at:str,expires_at:str,tenant_id:str?'), EntityDef(name='casbin_rule', fields_str='ptype:str,v0:str,v1:str,v2:str?,v3:str?,v4:str?,v5:str?,tenant_id:str?'), EntityDef(name='user_role', fields_str='user_id:str,role_id:str,assigned_at:str?,assigned_by:str?', depends_on=['user_id:auth_user', 'role_id:role']), EntityDef(name='role_permission', fields_str='role_id:str,permission_id:str', depends_on=['role_id:role', 'permission_id:permission']), EntityDef(name='user_permission', fields_str='user_id:str,permission_id:str,is_denied:bool=False', depends_on=['user_id:auth_user', 'permission_id:permission'])]), 'mfa': AuthModuleDef(name='mfa', entities=[EntityDef(name='mfa_method', fields_str='user_id:str,method_type:str,secret:str?,is_primary:bool=False,is_verified:bool=False,tenant_id:str?', enums=['method_type:totp|sms|email|webauthn'], depends_on=['user_id:auth_user']), EntityDef(name='mfa_backup_code', fields_str='user_id:str,code_hash:str,is_used:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='mfa_challenge', fields_str='user_id:str,method_type:str,challenge:str,expires_at:str,is_verified:bool=False,tenant_id:str?', depends_on=['user_id:auth_user'])]), 'api-keys': AuthModuleDef(name='api-keys', entities=[EntityDef(name='api_key', fields_str='user_id:str,name:str,key_hash:str,prefix:str,expires_at:str?,is_active:bool=True,last_used_at:str?,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='api_key_scope', fields_str='api_key_id:str,scope:str,tenant_id:str?', depends_on=['api_key_id:api_key'])]), 'oauth2': AuthModuleDef(name='oauth2', entities=[EntityDef(name='oauth2_client', fields_str='client_id:str,client_secret_hash:str,name:str,redirect_uris:str,grant_types:str,scopes:str,is_active:bool=True,tenant_id:str?'), EntityDef(name='oauth2_authorization_code', fields_str='client_id:str,user_id:str,code:str,redirect_uri:str,scopes:str,expires_at:str,is_used:bool=False,tenant_id:str?', depends_on=['user_id:auth_user']), EntityDef(name='oauth2_token', fields_str='client_id:str,user_id:str,access_token:str,refresh_token:str?,token_type:str=bearer,scopes:str,expires_at:str,tenant_id:str?', depends_on=['user_id:auth_user'])]), 'social': AuthModuleDef(name='social', entities=[EntityDef(name='social_provider', fields_str='name:str,client_id:str,client_secret:str,authorize_url:str,token_url:str,userinfo_url:str,scopes:str,is_active:bool=True,tenant_id:str?'), EntityDef(name='social_account', fields_str='user_id:str,provider_name:str,provider_user_id:str,access_token:str?,refresh_token:str?,tenant_id:str?', depends_on=['user_id:auth_user'])]), 'ldap': AuthModuleDef(name='ldap', entities=[EntityDef(name='ldap_config', fields_str='name:str,host:str,port:int=389,base_dn:str,bind_dn:str?,bind_password:str?,use_ssl:bool=False,is_active:bool=True,tenant_id:str?'), EntityDef(name='ldap_group_mapping', fields_str='ldap_group:str,role_id:str,tenant_id:str?', depends_on=['role_id:role'])]), 'claims': AuthModuleDef(name='claims', entities=[EntityDef(name='claim', fields_str='name:str,description:str?,claim_type:str,tenant_id:str?', enums=['claim_type:string|integer|boolean|json']), EntityDef(name='user_claim', fields_str='user_id:str,claim_id:str,value:str,tenant_id:str?', depends_on=['user_id:auth_user', 'claim_id:claim']), EntityDef(name='role_claim', fields_str='role_id:str,claim_id:str,value:str,tenant_id:str?', depends_on=['role_id:role', 'claim_id:claim'])]), 'audit': AuthModuleDef(name='audit', entities=[EntityDef(name='audit_log', fields_str='user_id:str?,action:str,resource:str,resource_id:str?,ip_address:str?,user_agent:str?,details:str?,tenant_id:str?')]), 'security': AuthModuleDef(name='security', entities=[EntityDef(name='tenant_security', fields_str='tenant_id:str,max_login_attempts:int=5,lockout_duration:int=900,password_min_length:int=8,require_mfa:bool=False,session_timeout:int=3600'), EntityDef(name='ip_rule', fields_str='ip_address:str,rule_type:str,description:str?,is_active:bool=True,tenant_id:str?', enums=['rule_type:whitelist|blacklist']), EntityDef(name='trusted_device', fields_str='user_id:str,device_hash:str,device_name:str?,last_used_at:str?,is_active:bool=True,tenant_id:str?', depends_on=['user_id:auth_user'])]), 'webhooks': AuthModuleDef(name='webhooks', entities=[EntityDef(name='webhook', fields_str='url:str,events:str,secret:str?,is_active:bool=True,last_triggered_at:str?,tenant_id:str?')])}
dataclass
¶
Python
EntityDef(name: str, fields_str: str, enums: list[str] = list(), depends_on: list[str] = list())
Definição de uma entidade RBAC.
cli¶
Comandos apogee auth:* — instalação de módulos RBAC via CLI global.
Registra auth:install e auth:list.
Source code in apogee_auth/cli.py
Python
def register(sub: argparse._SubParsersAction) -> None:
"""Registra auth:install e auth:list."""
p = sub.add_parser("auth:install", help="Instala módulos RBAC Auth")
p.add_argument("--modules", "-m", required=True, help="Módulos: core,mfa,api-keys")
p.add_argument("--domain", "-d", default="Auth", help="DomÃnio (default: Auth)")
p.add_argument("--context", "-c", default="main", help="Contexto (default: main)")
p.add_argument(
"--with-runtime",
action="store_true",
dest="with_runtime",
help="Renderiza AuthService/AuthController/AuthMiddleware (BL-003)",
)
p.add_argument(
"--package",
default=None,
help="Pacote Python alvo (default: derivado do contexto)",
)
p.set_defaults(func=_handle_install)
p = sub.add_parser("auth:uninstall", help="Remove módulos RBAC instalados (BL-005)")
p.add_argument("--modules", "-m", required=True, help="Módulos a remover")
p.add_argument("--context", "-c", default="main")
p.add_argument("--package", default=None)
p.add_argument(
"--dry-run",
action="store_true",
dest="dry_run",
help="Lista os arquivos sem remover",
)
p.add_argument(
"--rollback-migrations",
type=int,
default=0,
dest="rollback_migrations",
metavar="N",
help="Apoś remover, dispara apogee db-context-migrate-down N revisões",
)
p.set_defaults(func=_handle_uninstall)
p = sub.add_parser("auth:list", help="Lista módulos RBAC disponÃveis")
p.set_defaults(func=_handle_list)
template_renderer¶
Lightweight template renderer for apogee-auth.
Uses Jinja2 if available; falls back to a simple {{var}} substitution
otherwise so the package works out-of-the-box.
Render a single .j2 file. Falls back to simple substitution if Jinja2 isn't installed.
Source code in apogee_auth/template_renderer.py
Python
def render_template_file(template_path: Path, context: dict[str, Any]) -> str:
"""Render a single .j2 file. Falls back to simple substitution if Jinja2 isn't installed."""
source = template_path.read_text(encoding="utf-8")
try:
import jinja2 # type: ignore
env = jinja2.Environment(
undefined=jinja2.StrictUndefined,
keep_trailing_newline=True,
autoescape=False,
)
return env.from_string(source).render(**context)
except ImportError:
return _simple_render(source, context)
Python
render_runtime_files(target_root: Path, package: str, context_name: str | None = None) -> list[Path]
Render the auth runtime router into the framework's standard path.
Emits a single FastAPI router under
target_root/src/presentation/http/routers/auth_router.py which the
framework's FastAPI auto-discovery mounts at /api/v1/auth/*.
The router queries the AuthUserModel already generated by
auth:install in
src/infrastructure/persistence/db_context/<context>/sql/postgres/models/.
Returns list of files written.
Source code in apogee_auth/template_renderer.py
Python
def render_runtime_files(
target_root: Path,
package: str,
context_name: str | None = None,
) -> list[Path]:
"""Render the auth runtime router into the framework's standard path.
Emits a single FastAPI router under
``target_root/src/presentation/http/routers/auth_router.py`` which the
framework's FastAPI auto-discovery mounts at ``/api/v1/auth/*``.
The router queries the ``AuthUserModel`` already generated by
``auth:install`` in
``src/infrastructure/persistence/db_context/<context>/sql/postgres/models/``.
Returns list of files written.
"""
if not package:
raise ValueError("package cannot be empty")
ctx_name = context_name or package or "main"
templates_dir = Path(__file__).parent / "templates"
render_context = {"package": package, "context": ctx_name}
written: list[Path] = []
rendered = render_template_file(templates_dir / "auth_router.py.j2", render_context)
out = target_root / "src" / "presentation" / "http" / "routers" / "auth_router.py"
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(rendered, encoding="utf-8")
written.append(out)
# Ensure __init__.py exists in the routers package.
init_file = out.parent / "__init__.py"
if not init_file.exists():
init_file.write_text("", encoding="utf-8")
return written
uninstaller¶
Uninstaller for auth modules — reverses what auth:install generated.
Build a per-entity file list (entities + repositories + value_objects) honouring FK reverse order so that compound deletes don't break.
dataclass
¶
Python
build_plan(root: Path, modules: list[str], package: str, context: str) -> UninstallPlan
Plan removal of modules in reverse FK order.
Reverse FK order = entities with no dependents first... but in the
config, depends_on already encodes parent direction (child uses
parent_id:parent). So the simplest correct rule is: process entities
within a module in reverse order of declaration (children come
later in the lists), then process modules in reverse declaration.
Source code in apogee_auth/uninstaller.py
Python
def build_plan(
root: Path,
modules: list[str],
package: str,
context: str,
) -> UninstallPlan:
"""Plan removal of `modules` in reverse FK order.
Reverse FK order = entities with no dependents first... but in the
config, `depends_on` already encodes parent direction (child uses
`parent_id:parent`). So the simplest correct rule is: process entities
*within a module* in reverse order of declaration (children come
later in the lists), then process modules in reverse declaration.
"""
files: list[Path] = []
skipped: list[Path] = []
handled_modules: list[str] = []
for mod_name in reversed(modules):
if mod_name not in RBAC_MODULES:
continue
handled_modules.append(mod_name)
mod = RBAC_MODULES[mod_name]
for entity in reversed(mod.entities):
for path in _entity_file_paths(root, package, context, entity.name):
if path.exists():
files.append(path)
else:
skipped.append(path)
return UninstallPlan(files=files, skipped=skipped, modules=handled_modules)
Python
execute_plan(plan: UninstallPlan, dry_run: bool = False) -> dict
Apply the plan. With dry_run=True, lists what would happen
without touching disk.
Source code in apogee_auth/uninstaller.py
Python
def execute_plan(plan: UninstallPlan, dry_run: bool = False) -> dict:
"""Apply the plan. With ``dry_run=True``, lists what would happen
without touching disk."""
removed: list[str] = []
if dry_run:
return {
"dry_run": True,
"modules": list(plan.modules),
"would_remove": [str(p) for p in plan.files],
"skipped": [str(p) for p in plan.skipped[:20]], # cap noise
}
for path in plan.files:
try:
path.unlink()
removed.append(str(path))
except FileNotFoundError:
continue
except OSError as exc:
return {
"dry_run": False,
"error": f"could not remove {path}: {exc}",
"removed_so_far": removed,
}
# Tidy up empty directories left behind (best-effort)
for path in plan.files:
parent = path.parent
try:
if parent.exists() and not any(parent.iterdir()):
parent.rmdir()
except OSError:
pass
return {
"dry_run": False,
"modules": list(plan.modules),
"removed": removed,
}