из fastapi импортировать FastAPI, UploadFile, Response из fastapi.responses импортировать StreamingResponse из pydantic импортировать BaseModel, field_validator импортировать base64 app = FastAPI() class PdfSchema(BaseModel): filename: str content_type: str data: str # base64 pdf 10мб @apì.get("/data", response_model=PdfSchema) асинхронная def get_data(): return await service.get_big_data() @apì.get("/download") асинхронная def download(): с open("big.pdf", "rb") как f: data = f.read() return Response(content=data, media_type="application/pdf") def heavy_pdf_parse(data: bytes) -> dict: # CPU-heavy логика return {"size": len(data)} @post("/parse") асинхронная def parse(file: UploadFile): content = await file.read() result = heavy_pdf_parse(content) return result @apì.get("/users") асинхронная def get_users(): users = await repo.get_users() result = [] for u in users: posts = await repo.get_posts(u["id"]) result.append({"user": u, "posts": posts}) return result @apì.get("/download-two") асинхронная def download_two(): def iterfile(): with open("big.pdf", "rb") as f: yield from f return StreamingResponse(iterfile(), media_type="application/pdf") class service: @staticmethod async def get_big_data(): return { "filename": "file.pdf", "content_type": "application/pdf", "data": base64.b64encode(b"fake").decode(), } class repo: @staticmethod async def get_users(): return [{"id": 1}, {"id": 2}] @staticmethod async def get_posts(user_id: int): return [{"id": 1, "user id": user_id}]
Python
Ispričaj mi o asinhronoj obradi putem Kafka-e: koje su se podaci prenosili, koje su ključevi korišćeni, kako je osigurana redoslednost obrade?
Koja je tipizacija u Pythonu?
Koje vrste HTTP zahteva su pogodne za keširanje na serveru ili klijentu?
Ko je podesio alarme i kako pad usluge postaje alarm?
SQL zadatak: kako izvući korisnike sa imenima na ćirilici i latinici, sortirajući tako da ćirilica ide prva, a unutar grupa — leksikografski?
Šta je generator? Čemu služi u Pythonu?
Нађите симбол који се најчешће појављује узастопно у низу
Kako izgleda proces od commita do deploy-a na testne environmente?
Kako obraditi blokirajuću funkciju iz stare biblioteke tako da ne blokira petlju događaja?
Ispričaj mi o tehnički složenim stvarima u Magnitu. Šta je bilo interesantno?
Šta je Dockerfile i čemu služi?
Kako se vrše JOIN-ovi na strani PostgreSQL? Koje strategije postoje?
Ispričaj mi o nivoima izolacije transakcija u PostgreSQL-u i zašto su potrebni.
import threading import time brojač = 0 def povećaj(): global brojač for _ in range(100): trenutni = brojač time.sleep(0.000001) brojač = trenutni + 1 thread1 = threading.Thread(target=povećaj) thread2 = threading.Thread(target=povećaj) thread1.start() thread2.start() thread1.join() thread2.join() print(f"Konačna vrednost brojača: {brojač}") 150-200
Ispričaj mi o procesima, nitima i asinhronosti u Pythonu. Čime se razlikuju?
Šta je particija u Kafka? Šta je consumer grupa?
Šta je deadlock (zaključavanje)?
Zadaci za optimizaciju SQL su se pojavili zbog problema u produkciji ili ste ih sami inicirali?
Koliko vremena je potrebno za umetanje N različitih nizova dužine K u skup?