os.getenv -> config.py

aiws.py

@@ -1,6 +1,5 @@
 import asyncio
 import logging
-import os
 from contextlib import asynccontextmanager
 
 from aiogram import Bot, Dispatcher, Router, F
@@ -9,7 +8,6 @@ from aiogram.enums import ParseMode
 from aiogram.filters import CommandStart, Command
 from aiogram.types import Message
 from aiogram.fsm.storage.mongo import MongoStorage
-from dotenv import load_dotenv
 from fastapi import FastAPI
 from motor.motor_asyncio import AsyncIOMotorClient
 from prometheus_client import Info
@@ -17,6 +15,7 @@ from starlette.middleware.cors import CORSMiddleware
 from prometheus_fastapi_instrumentator import Instrumentator
 
 # --- ИМПОРТЫ ПРОЕКТА ---
+from config import settings
 from adapters.google_adapter import GoogleAdapter
 from adapters.s3_adapter import S3Adapter
 from api.service.generation_service import GenerationService
@@ -46,16 +45,17 @@ from api.endpoints.project_router import router as project_api_router
 from api.endpoints.idea_router import router as idea_api_router
 from api.endpoints.post_router import router as post_api_router
 
-load_dotenv()
+
 logger = logging.getLogger(__name__)
 
 # --- КОНФИГУРАЦИЯ ---
-BOT_TOKEN = os.getenv("BOT_TOKEN")
+# Настройки теперь берутся из config.py
-GEMINI_API_KEY = os.getenv("GEMINI_API_KEY")
+BOT_TOKEN = settings.BOT_TOKEN
+GEMINI_API_KEY = settings.GEMINI_API_KEY
 
-MONGO_HOST = os.getenv("MONGO_HOST")  # Например: mongodb://localhost:27017
+MONGO_HOST = settings.MONGO_HOST
-DB_NAME = os.getenv("DB_NAME", "my_bot_db")  # Имя базы данных
+DB_NAME = settings.DB_NAME
-ADMIN_ID = int(os.getenv("ADMIN_ID", 0))
+ADMIN_ID = settings.ADMIN_ID
 
 
 def setup_logging():
@@ -79,10 +79,10 @@ char_repo = CharacterRepo(mongo_client)
 
 # S3 Adapter
 s3_adapter = S3Adapter(
-    endpoint_url=os.getenv("MINIO_ENDPOINT", "http://localhost:9000"),
+    endpoint_url=settings.MINIO_ENDPOINT,
-    aws_access_key_id=os.getenv("MINIO_ACCESS_KEY", "minioadmin"),
+    aws_access_key_id=settings.MINIO_ACCESS_KEY,
-    aws_secret_access_key=os.getenv("MINIO_SECRET_KEY", "minioadmin"),
+    aws_secret_access_key=settings.MINIO_SECRET_KEY,
-    bucket_name=os.getenv("MINIO_BUCKET", "ai-char")
+    bucket_name=settings.MINIO_BUCKET
 )
 
 dao = DAO(mongo_client, s3_adapter)  # Главный DAO для бота
@@ -259,7 +259,7 @@ if __name__ == "__main__":
     async def main():
         # Создаем конфигурацию uvicorn вручную
         # loop="asyncio" заставляет использовать стандартный цикл
-        config = uvicorn.Config(app, host="0.0.0.0", port=8090, loop="asyncio", timeout_keep_alive=120, env_file=".env.development")
+        config = uvicorn.Config(app, host="0.0.0.0", port=8090, loop="asyncio", timeout_keep_alive=120)
         server = uvicorn.Server(config)
 
         # Запускаем сервер (lifespan запустится внутри)

api/endpoints/generation_router.py

@@ -8,6 +8,7 @@ from fastapi.params import Depends
 from starlette import status
 from starlette.requests import Request
 
+from config import settings
 from api.dependency import get_generation_service, get_project_id, get_dao
 from api.endpoints.auth import get_current_user
 from api.models import (
@@ -197,7 +198,7 @@ async def import_external_generation(
     body = await request.body()
     
     # Verify signature
-    secret = os.getenv("EXTERNAL_API_SECRET")
+    secret = settings.EXTERNAL_API_SECRET
     if not secret:
         logger.error("EXTERNAL_API_SECRET not configured")
         raise HTTPException(status_code=500, detail="Server configuration error")

requirements.txt

@@ -51,3 +51,4 @@ python-jose[cryptography]==3.3.0
 python-multipart==0.0.22
 email-validator
 prometheus-fastapi-instrumentator
+pydantic-settings==2.13.0

tests/test_character_crud.py

@@ -3,17 +3,17 @@ import pytest
 from fastapi.testclient import TestClient
 from unittest.mock import MagicMock
 from motor.motor_asyncio import AsyncIOMotorClient
-import os
 import asyncio
+from config import settings
 
-from main import app
+from aiws import app
 from api.endpoints.auth import get_current_user
 from api.dependency import get_dao
 from repos.dao import DAO
 from models.Character import Character
 
 # Config for test DB
-MONGO_HOST = os.getenv("MONGO_HOST", "mongodb://admin:super_secure_password@31.59.58.220:27017")
+MONGO_HOST = settings.MONGO_HOST
 DB_NAME = "bot_db_test_chars"
 
 # Mock User

tests/test_external_import.py

@@ -10,13 +10,13 @@ import json
 import requests
 import base64
 import os
-from dotenv import load_dotenv
+from config import settings
 
-load_dotenv()
+# Load env is not needed as settings handles it
 
 # Configuration
 API_URL = "http://localhost:8090/api/generations/import"
-SECRET = os.getenv("EXTERNAL_API_SECRET", "your_super_secret_key_change_this_in_production")
+SECRET = settings.EXTERNAL_API_SECRET or "your_super_secret_key_change_this_in_production"
 
 # Sample generation data
 generation_data = {

tests/test_idea.py

@@ -10,11 +10,10 @@ from repos.dao import DAO
 from models.Idea import Idea
 from models.Generation import Generation, GenerationStatus
 from models.enums import AspectRatios, Quality
+from config import settings
 
-load_dotenv()
+MONGO_HOST = settings.MONGO_HOST
-
+DB_NAME = settings.DB_NAME
-MONGO_HOST = os.getenv("MONGO_HOST", "mongodb://localhost:27017")
-DB_NAME = os.getenv("DB_NAME", "bot_db")
 
 print(f"Connecting to MongoDB: {MONGO_HOST}, DB: {DB_NAME}")
 

tests/test_s3_connection.py

@@ -1,15 +1,14 @@
 import asyncio
 import os
-from dotenv import load_dotenv
+from config import settings
 from adapters.s3_adapter import S3Adapter
 
 async def test_s3():
-    load_dotenv()
     
-    endpoint = os.getenv("MINIO_ENDPOINT", "http://localhost:9000")
+    endpoint = settings.MINIO_ENDPOINT
-    access_key = os.getenv("MINIO_ACCESS_KEY")
+    access_key = settings.MINIO_ACCESS_KEY
-    secret_key = os.getenv("MINIO_SECRET_KEY")
+    secret_key = settings.MINIO_SECRET_KEY
-    bucket = os.getenv("MINIO_BUCKET")
+    bucket = settings.MINIO_BUCKET
     
     print(f"Connecting to {endpoint}, bucket: {bucket}")
     

tests/test_scheduler.py

@@ -4,13 +4,11 @@ from datetime import datetime, timedelta, UTC
 from motor.motor_asyncio import AsyncIOMotorClient
 from models.Generation import Generation, GenerationStatus
 from repos.generation_repo import GenerationRepo
-from dotenv import load_dotenv
+from config import settings
-
-load_dotenv()
 
 # Mock configs if not present in env
-MONGO_HOST = os.getenv("MONGO_HOST", "mongodb://localhost:27017")
+MONGO_HOST = settings.MONGO_HOST
-DB_NAME = os.getenv("DB_NAME", "bot_db")
+DB_NAME = settings.DB_NAME
 
 print(f"Connecting to MongoDB: {MONGO_HOST}, DB: {DB_NAME}")
 

tests/verify_albums_manual.py

@@ -10,10 +10,11 @@ from repos.dao import DAO
 from models.Album import Album
 from models.Generation import Generation, GenerationStatus
 from models.enums import AspectRatios, Quality
+from config import settings
 
 # Mock config
 # Use the same host as aiws.py but different DB
-MONGO_HOST = os.getenv("MONGO_HOST", "mongodb://admin:super_secure_password@31.59.58.220:27017")
+MONGO_HOST = settings.MONGO_HOST
 DB_NAME = "bot_db_test_albums"
 
 async def test_albums():
@@ -83,8 +84,6 @@ async def test_albums():
         client.close()
 
 if __name__ == "__main__":
-    from dotenv import load_dotenv
-    load_dotenv()
     try:
         asyncio.run(test_albums())
     except Exception as e:

tests/verify_minio_integration.py

@@ -1,29 +1,28 @@
 import asyncio
 import os
 from datetime import datetime
-from dotenv import load_dotenv
 from motor.motor_asyncio import AsyncIOMotorClient
 
+from config import settings
 from models.Asset import Asset, AssetType
 from repos.assets_repo import AssetsRepo
 from adapters.s3_adapter import S3Adapter
 
-# Load env to get credentials
+# Load env is not needed as settings handles it
-load_dotenv()
 
 async def test_integration():
     print("🚀 Starting integration test...")
     
     # 1. Setup Dependencies
-    mongo_uri = os.getenv("MONGO_HOST", "mongodb://localhost:27017")
+    mongo_uri = settings.MONGO_HOST
     client = AsyncIOMotorClient(mongo_uri)
-    db_name = os.getenv("DB_NAME", "bot_db_test")
+    db_name = settings.DB_NAME + "_test"
     
     s3_adapter = S3Adapter(
-        endpoint_url=os.getenv("MINIO_ENDPOINT", "http://localhost:9000"),
+        endpoint_url=settings.MINIO_ENDPOINT,
-        aws_access_key_id=os.getenv("MINIO_ACCESS_KEY", "admin"),
+        aws_access_key_id=settings.MINIO_ACCESS_KEY,
-        aws_secret_access_key=os.getenv("MINIO_SECRET_KEY", "SuperSecretPassword123!"),
+        aws_secret_access_key=settings.MINIO_SECRET_KEY,
-        bucket_name=os.getenv("MINIO_BUCKET", "ai-char")
+        bucket_name=settings.MINIO_BUCKET
     )
     
     repo = AssetsRepo(client, s3_adapter, db_name=db_name)

utils/security.py

@@ -3,12 +3,12 @@ from typing import Optional, Union, Any
 
 from jose import jwt
 from passlib.context import CryptContext
+from config import settings
 
-# Настройки безопасности (лучше вынести в config/env, но для старта здесь)
+# Настройки безопасности берутся из config.py
-# SECRET_KEY должен быть сложным и секретным в продакшене!
+SECRET_KEY = settings.SECRET_KEY
-SECRET_KEY = "CHANGE_ME_TO_A_SUPER_SECRET_KEY" 
+ALGORITHM = settings.ALGORITHM
-ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = settings.ACCESS_TOKEN_EXPIRE_MINUTES
-ACCESS_TOKEN_EXPIRE_MINUTES = 30 * 24 * 60  # 30 дней, например
 
 pwd_context = CryptContext(schemes=["argon2"], deprecated="auto")