mascai's questions

我正在尝试在我的 FastAPI 应用程序中使用 keycloak 我的代码

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import jwt, JWTError
from keycloak import KeycloakOpenID
import requests
import logging
import os

from .config import settings


# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

# Keycloak configuration
KEYCLOAK_SERVER_URL = settings.KEYCLOAK_SERVER_URL
KEYCLOAK_REALM = settings.KEYCLOAK_REALM
KEYCLOAK_CLIENT_ID = settings.KEYCLOAK_CLIENT_ID
KEYCLOAK_CLIENT_SECRET = settings.KEYCLOAK_CLIENT_SECRET
ALGORITHM = "RS256"
TOKEN_URL = f"{KEYCLOAK_SERVER_URL}/realms/fastapi-realm/protocol/openid-connect/token"


# Initialize KeycloakOpenID
keycloak_openid = KeycloakOpenID(
    server_url=f"{KEYCLOAK_SERVER_URL}",
    client_id=KEYCLOAK_CLIENT_ID,
    realm_name=KEYCLOAK_REALM,
    client_secret_key=KEYCLOAK_CLIENT_SECRET,
    verify=False
)
config_well_known = keycloak_openid.well_known()

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

def verify_token(token: str = Depends(oauth2_scheme)):
    try:
        decoded_token = keycloak_openid.decode_token(
            token,validate=True,
        )
        username = decoded_token['preferred_username']
        logger.info(f"Decoded token: {decoded_token}")
        # Verify the issuer claim
        issuer = decoded_token["iss"]

        expected_issuer = f"{KEYCLOAK_SERVER_URL}/realms/{KEYCLOAK_REALM}"
        I# Token example -- token: {'exp': 1731303036, 'iat': 1731267036, 'jti': 'f1b71d25-4de6-4c03-b5f5-d9726b39d51f', 'iss': 'https://feast-keycloak.pimc-st.innodev.local/realms/feast-realm', 'aud': 'account', 'sub': 'ac48f45e-f26b-4380-bde8-e752febb6d18', 'typ': 'Bearer', 'azp': 'feast-client-id', 'session_state': 'b36cd197-247d-447e-9f3d-6cf1fecae7d6', 'acr': '1', 'allowed-origins': ['https://feast-frontend.pimc-st.innodev.local', '/*', 'http://localhost:5173'], 'realm_access': {'roles': ['default-roles-feast-realm', 'offline_access', 'uma_authorization']}, 'resource_access': {'account': {'roles': ['manage-account', 'manage-account-links', 'view-profile']}}, 'scope': 'profile email', 'sid': 'b36cd197-247d-447e-9f3d-6cf1fecae7d6', 'email_verified': False, 'name': 'A B', 'preferred_username': 'my_username', 'given_name': 'A', 'family_name': 'B', 'email': '[email protected]'}
        logger.info(f"XXX_ issuer={issuer}")
        logger.info(f"XXX_ expected_issuer={expected_issuer}")
        if issuer != expected_issuer:
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid issuer")
        
        logger.info(f"username: {username}")
        return decoded_token
    except Exception as e:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")

在我的 main.py 中我有以下代码

from fastapi import Depends, FastAPI, HTTPException, status, Security


from .keycloak import verify_token, oauth2_scheme, KEYCLOAK_CLIENT_ID, KEYCLOAK_CLIENT_SECRET, TOKEN_URL

app = FastAPI()

@app.post("/project", response_model=schemas.Project, tags=["project",])
def create_project(project: schemas.CreateProject, db: Session = Depends(get_db), payload: dict = Security(verify_token)):
    ...


@app.post("/login")
def get_token(body: schemas.Login):
    data = {
        "grant_type": "password", # TODO: clarify grant_type client_credentials (requires only client id and secret or password - requires password and login)
        "client_id": KEYCLOAK_CLIENT_ID,
        "client_secret": KEYCLOAK_CLIENT_SECRET,
        "password": body.password,
        "username": body.login
    }
    response = requests.post(TOKEN_URL, data=data,  verify=False)
    return JSONResponse(status_code=response.status_code, content=response.json())

我通过 /login 方法获取令牌然后我像这样应用令牌：

对于请求/project我有一个错误：

“无效令牌”

如何修复错误？

我正在构建 fastapi + sqlalchemy 应用程序

我的代码

sql.py

form sqlalchemy import Table, String, BigInteger, MetaData

metadata = MetaData()

custom_datasets = Table(
    "custom_datasets",
    metadata,
    Column("project_id", String(500), primary_key=True),
    Column("id", BigInteger, primary_key=True),
    Column("name", String(500), unique=True),
    Column("path", String(500), unique=True),
    Column("feature_service", String(500), nullable=False),
    Column("last_updated_timestamp", BigInteger, nullable=False),
)

主程序

import os
from sqlalchemy.orm import Session
from sqlalchemy import select

def get_db():
    if os.getenv("REGISTRY_CONNECTION_STRING"):
        db = SessionLocal()
        yield db
    yield None


@app.get("/datasets/")
def get_project_registry(body: Dict[str, Any], db: Session = Depends(get_db)):
    print("XXX_ ", custom_datasets.c) # XXX_  ImmutableColumnCollection(custom_datasets.project_id, custom_datasets.id, 
    stmt = select(custom_datasets).where(
        custom_datasets.c.project_id in body.project_ids
    )
    res = db.execute().all()
    return res

我有一个错误

python3.8/site-packages/feast/feature_server.py", line 469, in get_project_registry
    custom_datasets.c.project_id in body.project_ids
AttributeError: 'dict' object has no attribute 'project_ids'

如何修复错误？

更新：

过滤对请求不起作用

GET http://127.0.0.1:8009/datasets
{
    "project_ids": ["proj_id1"]
}

表格如下所示：

testdb=# SELECT * FROM  custom_datasets;
 project_id | id | name  |   path    | feature_service | last_updated_timestamp 
------------+----+-------+-----------+-----------------+------------------------
 proj_id1   |  1 | name1 | path/path | service_name    |                      1

更新代码

    @app.get("/datasets/")
    def get_project_registry(body: Dict[str, Any], db: Session = Depends(get_db)):
        print("XXX_ ", body['project_ids'])  # prints XXX_  ['proj_id1']
        stmt = select(custom_datasets).where(
            custom_datasets.c.project_id in body['project_ids']
        )
        #stmt = select(custom_datasets)
        res = db.execute(stmt).all()
        return res

我正在创建 FastApi + KeyCloak 应用程序。已创建领域、客户端和用户

客户端配置

我的用户配置

我可以借助此请求获取令牌

但是对于 grant_typeclient-credentials我有一个错误：

{
    "error": "unsupported_grant_type",
    "error_description": "Unsupported grant_type"
}

密码和客户端凭证有什么区别？如何支持客户端凭证？

我想以以下格式设置 crontab 计划，0 19-8 * * * 但出现错误，提示不支持此格式

我想每小时运行一次 19、20、21、22、23、24、0、1、... 8

我知道这种格式是可行的 0 19,20,21,22,23,0,1,2,3,4,5,6,7,8 * * *，但是在我的例子中我该如何设置范围？

我想连接多个表中的列。对于一个表，连接工作正常

select org.catalog_truck.plate_number as "Госномер грузовика",
shift_id as "ID смены",
trip_id as "ID рейса",
*
from core.monitoring_stages
join org.catalog_truck
on vehicle_id = org.catalog_truck.catalog_truck_id
where gen_id = (SELECT max(gen_id)
FROM core.monitoring_tripmachine_generation
WHERE ready=true and timerange_from > '2024-09-23 00:00:00.000 +0300' and timerange_to < '2024-09-23 23:59:59.000 +0300')
ORDER by dtstart

但是对于两个连接（以及前两行中的两个选择，我有一个错误）

select org.catalog_truck.plate_number as "Госномер грузовика",
select org.catalog_field.name as "Поле погрузки",   <--- select for the second join (error in this line)
shift_id as "ID смены",
trip_id as "ID рейса",
*
from core.monitoring_stages
join org.catalog_truck
on vehicle_id = org.catalog_truck.catalog_truck_id
where gen_id = (SELECT max(gen_id)
from core.monitoring_stages <--- second join
join org.catalog_field cf 
on field_id = org.catalog_field.name
FROM core.monitoring_tripmachine_generation
WHERE ready=true and timerange_from > '2024-09-23 00:00:00.000 +0300' and timerange_to < '2024-09-23 23:59:59.000 +0300')
ORDER by dtstart

错误文本：

SQL 错误 [42601]：错误：“select”处或附近的语法错误 位置：66

错误位置：行：2 位置：65

如何对两个连接应用两个选择？

我正在尝试在 MacOS 上运行 userver（https://userver.tech/ ）

我正在关注这个教程：https://userver.tech/d3/da9/md_en_2userver_2tutorial_2build.html

我的行动：

git clone --depth 1 https://github.com/userver-framework/service_template.git && \
git clone --depth 1 https://github.com/userver-framework/userver.git service_template/third_party/userver && \
cd service_template

brew install $(cat third_party/userver/scripts/docs/en/deps/macos.md | tr '\n' ' ')


make build-release && \
make service-start-release

我有一个错误：

 25%] Building CXX object third_party/userver/universal/CMakeFiles/userver-universal.dir/src/utils/impl/byte_utils.cpp.o
/Users/mascai/root_folder/dev/projects/49_USERVER/01_tutorial/service_template/third_party/userver/universal/src/utils/boost_uuid7.cpp:49:27: error: no matching conversion for functional-style cast from 'boost::uuids::uuid::data_type' to 'utils::span<std::uint8_t>' (aka 'span<unsigned char>')
      GenerateRandomBlock(utils::span<std::uint8_t>(uuid.data).subspan(8));
                          ^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~

我猜问题与 Boost 版本有关。我的 Boost 版本是 1.86.0，clang 是 13.0.0。

如何跳过循环中的最后一个字符串元素？

我的解决方案运行良好

fn main() {
    let s = "123456".to_string();
    for ch in (&s[..s.len()-1]).chars() { // skip the last element
        println!("{}", ch);
    }
}

但还有更好的吗？

类似于skip() https://doc.rust-lang.org/std/iter/trait.Iterator.html#method.skip 允许跳过循环中的第一个元素

我想在 Rust 中迭代两个字符串。我读到使用索引（nth()方法）效率不高。这就是我使用迭代器的原因

fn main() {
    let s1 = String::from("abacde");
    let s2 = String::from("123456");
    let mut iter1 = s1.chars();
    let mut iter2 = s2.chars();

    while let Some(ch1) = iter1.next() && let Some(ch2) = iter2.next(){
        println!("{:?}", ch1);
    }
}

我有一个错误

错误[E0658]：let该位置的表达式不稳定

如何修复它？